Refactor MAM catchup. Fetch from latest to earliest message.

8 files changed, 401 insertions, 119 deletions

diff --git a/libdino/src/service/conversation_manager.vala b/libdino/src/service/conversation_manager.vala
index 1ba53b35..c473ea77 100644
--- a/libdino/src/service/conversation_manager.vala
+++ b/libdino/src/service/conversation_manager.vala
@@ -163,7 +163,7 @@ public class ConversationManager : StreamInteractionModule, Object {
 
             if (stanza != null) {
                 bool is_mam_message = Xep.MessageArchiveManagement.MessageFlag.get_flag(stanza) != null;
-                bool is_recent = message.local_time.compare(new DateTime.now_utc().add_hours(-24)) > 0;
+                bool is_recent = message.local_time.compare(new DateTime.now_utc().add_days(-3)) > 0;
                 if (is_mam_message && !is_recent) return false;
             }
             stream_interactor.get_module(ConversationManager.IDENTITY).start_conversation(conversation);
diff --git a/libdino/src/service/database.vala b/libdino/src/service/database.vala
index 54854fd1..e230c3af 100644
--- a/libdino/src/service/database.vala
+++ b/libdino/src/service/database.vala
@@ -7,7 +7,7 @@ using Dino.Entities;
 namespace Dino {
 
 public class Database : Qlite.Database {
-    private const int VERSION = 10;
+    private const int VERSION = 11;
 
     public class AccountTable : Table {
         public Column<int> id = new Column.Integer("id") { primary_key = true, auto_increment = true };
@@ -185,6 +185,21 @@ public class Database : Qlite.Database {
         }
     }
 
+    public class MamCatchupTable : Table {
+        public Column<int> id = new Column.Integer("id") { primary_key = true, auto_increment = true };
+        public Column<int> account_id = new Column.Integer("account_id") { not_null = true };
+        public Column<bool> from_end = new Column.BoolInt("from_end");
+        public Column<string> from_id = new Column.Text("from_id");
+        public Column<long> from_time = new Column.Long("from_time") { not_null = true };
+        public Column<string> to_id = new Column.Text("to_id");
+        public Column<long> to_time = new Column.Long("to_time") { not_null = true };
+
+        internal MamCatchupTable(Database db) {
+            base(db, "mam_catchup");
+            init({id, account_id, from_end, from_id, from_time, to_id, to_time});
+        }
+    }
+
     public class SettingsTable : Table {
         public Column<int> id = new Column.Integer("id") { primary_key = true, auto_increment = true };
         public Column<string> key = new Column.Text("key") { unique = true, not_null = true };
@@ -206,6 +221,7 @@ public class Database : Qlite.Database {
     public AvatarTable avatar { get; private set; }
     public EntityFeatureTable entity_feature { get; private set; }
     public RosterTable roster { get; private set; }
+    public MamCatchupTable mam_catchup { get; private set; }
     public SettingsTable settings { get; private set; }
 
     public Map<int, Jid> jid_table_cache = new HashMap<int, Jid>();
@@ -224,8 +240,9 @@ public class Database : Qlite.Database {
         avatar = new AvatarTable(this);
         entity_feature = new EntityFeatureTable(this);
         roster = new RosterTable(this);
+        mam_catchup = new MamCatchupTable(this);
         settings = new SettingsTable(this);
-        init({ account, jid, content_item, message, real_jid, file_transfer, conversation, avatar, entity_feature, roster, settings });
+        init({ account, jid, content_item, message, real_jid, file_transfer, conversation, avatar, entity_feature, roster, mam_catchup, settings });
         try {
             exec("PRAGMA synchronous=0");
         } catch (Error e) { }
@@ -280,6 +297,15 @@ public class Database : Qlite.Database {
                 error("Failed to upgrade to database version 9: %s", e.message);
             }
         }
+        if (oldVersion < 11) {
+            try {
+                exec("""
+                insert into mam_catchup (account_id, from_end, from_time, to_time)
+                select id, 1, 0, mam_earliest_synced from account where mam_earliest_synced not null and mam_earliest_synced > 0""");
+            } catch (Error e) {
+                error("Failed to upgrade to database version 11: %s", e.message);
+            }
+        }
     }
 
     public ArrayList<Account> get_accounts() {
@@ -373,47 +399,6 @@ public class Database : Qlite.Database {
         return ret;
     }
 
-    public bool contains_message(Message query_message, Account account) {
-        QueryBuilder builder = message.select()
-                .with(message.account_id, "=", account.id)
-                .with(message.counterpart_id, "=", get_jid_id(query_message.counterpart))
-                .with(message.body, "=", query_message.body)
-                .with(message.time, "<", (long) query_message.time.add_minutes(1).to_unix())
-                .with(message.time, ">", (long) query_message.time.add_minutes(-1).to_unix());
-        if (query_message.stanza_id != null) {
-            builder.with(message.stanza_id, "=", query_message.stanza_id);
-        } else {
-            builder.with_null(message.stanza_id);
-        }
-        if (query_message.counterpart.resourcepart != null) {
-            builder.with(message.counterpart_resource, "=", query_message.counterpart.resourcepart);
-        } else {
-            builder.with_null(message.counterpart_resource);
-        }
-        return builder.count() > 0;
-    }
-
-    public bool contains_message_by_stanza_id(Message query_message, Account account) {
-        QueryBuilder builder =  message.select()
-                .with(message.stanza_id, "=", query_message.stanza_id)
-                .with(message.counterpart_id, "=", get_jid_id(query_message.counterpart))
-                .with(message.account_id, "=", account.id);
-        if (query_message.counterpart.resourcepart != null) {
-            builder.with(message.counterpart_resource, "=", query_message.counterpart.resourcepart);
-        } else {
-            builder.with_null(message.counterpart_resource);
-        }
-        return builder.count() > 0;
-    }
-
-    public bool contains_message_by_server_id(Account account, Jid counterpart, string server_id) {
-        QueryBuilder builder =  message.select()
-                .with(message.server_id, "=", server_id)
-                .with(message.counterpart_id, "=", get_jid_id(counterpart))
-                .with(message.account_id, "=", account.id);
-        return builder.count() > 0;
-    }
-
     public Message? get_message_by_id(int id) {
         Row? row = message.row_with(message.id, id).inner;
         if (row != null) {
diff --git a/libdino/src/service/message_processor.vala b/libdino/src/service/message_processor.vala
index 604578b5..98239eb8 100644
--- a/libdino/src/service/message_processor.vala
+++ b/libdino/src/service/message_processor.vala
@@ -1,7 +1,9 @@
 using Gee;
 
 using Xmpp;
+using Xmpp.Xep;
 using Dino.Entities;
+using Qlite;
 
 namespace Dino {
 
@@ -20,6 +22,11 @@ public class MessageProcessor : StreamInteractionModule, Object {
     private StreamInteractor stream_interactor;
     private Database db;
     private Object lock_send_unsent;
+    private HashMap<Account, int> current_catchup_id = new HashMap<Account, int>(Account.hash_func, Account.equals_func);
+    private HashMap<Account, HashMap<string, DateTime>> mam_times = new HashMap<Account, HashMap<string, DateTime>>();
+    public HashMap<string, int> hitted_range = new HashMap<string, int>();
+    public HashMap<Account, string> catchup_until_id = new HashMap<Account, string>(Account.hash_func, Account.equals_func);
+    public HashMap<Account, DateTime> catchup_until_time = new HashMap<Account, DateTime>(Account.hash_func, Account.equals_func);
 
     public static void start(StreamInteractor stream_interactor, Database db) {
         MessageProcessor m = new MessageProcessor(stream_interactor, db);
@@ -29,14 +36,23 @@ public class MessageProcessor : StreamInteractionModule, Object {
     private MessageProcessor(StreamInteractor stream_interactor, Database db) {
         this.stream_interactor = stream_interactor;
         this.db = db;
+
+        received_pipeline.connect(new DeduplicateMessageListener(this, db));
+        received_pipeline.connect(new FilterMessageListener());
+        received_pipeline.connect(new StoreMessageListener(stream_interactor));
+        received_pipeline.connect(new MamMessageListener(stream_interactor));
+
         stream_interactor.account_added.connect(on_account_added);
+
         stream_interactor.connection_manager.connection_state_changed.connect((account, state) => {
             if (state == ConnectionManager.ConnectionState.CONNECTED) send_unsent_messages(account);
         });
-        received_pipeline.connect(new DeduplicateMessageListener(db));
-        received_pipeline.connect(new FilterMessageListener());
-        received_pipeline.connect(new StoreMessageListener(stream_interactor));
-        received_pipeline.connect(new MamMessageListener(stream_interactor));
+
+        stream_interactor.connection_manager.stream_opened.connect((account, stream) => {
+            debug("MAM: [%s] Reset catchup_id", account.bare_jid.to_string());
+            current_catchup_id.unset(account);
+            mam_times[account] = new HashMap<string, DateTime>();
+        });
     }
 
     public Entities.Message send_text(string text, Conversation conversation) {
@@ -65,22 +81,227 @@ public class MessageProcessor : StreamInteractionModule, Object {
         stream_interactor.module_manager.get_module(account, Xmpp.MessageModule.IDENTITY).received_message.connect( (stream, message) => {
             on_message_received.begin(account, message);
         });
+        XmppStream? stream_bak = null;
         stream_interactor.module_manager.get_module(account, Xmpp.Xep.MessageArchiveManagement.Module.IDENTITY).feature_available.connect( (stream) => {
-            DateTime start_time = account.mam_earliest_synced.to_unix() > 60 ? account.mam_earliest_synced.add_minutes(-1) : account.mam_earliest_synced;
-            stream.get_module(Xep.MessageArchiveManagement.Module.IDENTITY).query_archive(stream, null, start_time, null, () => {
-                history_synced(account);
-            });
+            if (stream == stream_bak) return;
+
+            current_catchup_id.unset(account);
+            stream_bak = stream;
+            debug("MAM: [%s] MAM available", account.bare_jid.to_string());
+            do_mam_catchup.begin(account);
+        });
+
+        stream_interactor.module_manager.get_module(account, Xmpp.MessageModule.IDENTITY).received_message_unprocessed.connect((stream, message) => {
+            if (!message.from.equals(account.bare_jid)) return;
+
+            Xep.MessageArchiveManagement.Flag? mam_flag = stream != null ? stream.get_flag(Xep.MessageArchiveManagement.Flag.IDENTITY) : null;
+            if (mam_flag == null) return;
+            string? id = message.stanza.get_deep_attribute(mam_flag.ns_ver + ":result", "id");
+            if (id == null) return;
+            StanzaNode? delay_node = message.stanza.get_deep_subnode(mam_flag.ns_ver + ":result", "urn:xmpp:forward:0:forwarded", "urn:xmpp:delay:delay");
+            if (delay_node == null) return;
+            DateTime? time = DelayedDelivery.Module.get_time_for_node(delay_node);
+            if (time == null) return;
+            mam_times[account][id] = time;
+
+            string? query_id = message.stanza.get_deep_attribute(mam_flag.ns_ver + ":result", mam_flag.ns_ver + ":queryid");
+            if (query_id != null && id == catchup_until_id[account]) {
+                debug("MAM: [%s] Hitted range (id) %s", account.bare_jid.to_string(), id);
+                hitted_range[query_id] = -2;
+            }
         });
     }
 
+    private async void do_mam_catchup(Account account) {
+        debug("MAM: [%s] Start catchup", account.bare_jid.to_string());
+        string? earliest_id = null;
+        DateTime? earliest_time = null;
+        bool continue_sync = true;
+
+        while (continue_sync) {
+            continue_sync = false;
+
+            // Get previous row
+            var previous_qry = db.mam_catchup.select().with(db.mam_catchup.account_id, "=", account.id).order_by(db.mam_catchup.to_time, "DESC");
+            if (current_catchup_id.has_key(account)) {
+                previous_qry.with(db.mam_catchup.id, "!=", current_catchup_id[account]);
+            }
+            RowOption previous_row = previous_qry.single().row();
+            if (previous_row.is_present()) {
+                catchup_until_id[account] = previous_row[db.mam_catchup.to_id];
+                catchup_until_time[account] = (new DateTime.from_unix_utc(previous_row[db.mam_catchup.to_time])).add_minutes(-5);
+                debug("MAM: [%s] Previous entry exists", account.bare_jid.to_string());
+            } else {
+                catchup_until_id.unset(account);
+                catchup_until_time.unset(account);
+            }
+
+            string query_id = Xmpp.random_uuid();
+            yield get_mam_range(account, query_id, null, null, earliest_time, earliest_id);
+
+            if (!hitted_range.has_key(query_id)) {
+                debug("MAM: [%s] Set catchup end reached", account.bare_jid.to_string());
+                db.mam_catchup.update()
+                    .set(db.mam_catchup.from_end, true)
+                    .with(db.mam_catchup.id, "=", current_catchup_id[account])
+                    .perform();
+            }
+
+            if (hitted_range.has_key(query_id)) {
+                if (merge_ranges(account, null)) {
+                    RowOption current_row = db.mam_catchup.row_with(db.mam_catchup.id, current_catchup_id[account]);
+                    bool range_from_complete = current_row[db.mam_catchup.from_end];
+                    if (!range_from_complete) {
+                        continue_sync = true;
+                        earliest_id = current_row[db.mam_catchup.from_id];
+                        earliest_time = (new DateTime.from_unix_utc(current_row[db.mam_catchup.from_time])).add_seconds(1);
+                    }
+                }
+            }
+        }
+    }
+
+    /*
+     * Merges the row with `current_catchup_id` with the previous range (optional: with `earlier_id`)
+     * Changes `current_catchup_id` to the previous range
+     */
+    private bool merge_ranges(Account account, int? earlier_id) {
+        RowOption current_row = db.mam_catchup.row_with(db.mam_catchup.id, current_catchup_id[account]);
+        RowOption previous_row = null;
+
+        if (earlier_id != null) {
+            previous_row = db.mam_catchup.row_with(db.mam_catchup.id, earlier_id);
+        } else {
+            previous_row = db.mam_catchup.select()
+                .with(db.mam_catchup.account_id, "=", account.id)
+                .with(db.mam_catchup.id, "!=", current_catchup_id[account])
+                .order_by(db.mam_catchup.to_time, "DESC").single().row();
+        }
+
+        if (!previous_row.is_present()) {
+            debug("MAM: [%s] Merging: No previous row", account.bare_jid.to_string());
+            return false;
+        }
+
+        var qry = db.mam_catchup.update().with(db.mam_catchup.id, "=", previous_row[db.mam_catchup.id]);
+        debug("MAM: [%s] Merging %ld-%ld with %ld- %ld", account.bare_jid.to_string(), previous_row[db.mam_catchup.from_time], previous_row[db.mam_catchup.to_time], current_row[db.mam_catchup.from_time], current_row[db.mam_catchup.to_time]);
+        if (current_row[db.mam_catchup.from_time] < previous_row[db.mam_catchup.from_time]) {
+            qry.set(db.mam_catchup.from_id, current_row[db.mam_catchup.from_id])
+                    .set(db.mam_catchup.from_time, current_row[db.mam_catchup.from_time]);
+        }
+        if (current_row[db.mam_catchup.to_time] > previous_row[db.mam_catchup.to_time]) {
+            qry.set(db.mam_catchup.to_id, current_row[db.mam_catchup.to_id])
+                .set(db.mam_catchup.to_time, current_row[db.mam_catchup.to_time]);
+        }
+        qry.perform();
+
+        current_catchup_id[account] = previous_row[db.mam_catchup.id];
+
+        db.mam_catchup.delete().with(db.mam_catchup.id, "=", current_row[db.mam_catchup.id]).perform();
+
+        return true;
+    }
+
+    private async bool get_mam_range(Account account, string? query_id, DateTime? from_time, string? from_id, DateTime? to_time, string? to_id) {
+        debug("MAM: [%s] Get range %s - %s", account.bare_jid.to_string(), from_time != null ? from_time.to_string() : "", to_time != null ? to_time.to_string() : "");
+        XmppStream stream = stream_interactor.get_stream(account);
+
+        Iq.Stanza? iq = yield stream.get_module(Xep.MessageArchiveManagement.Module.IDENTITY).query_archive(stream, null, query_id, from_time, from_id, to_time, to_id);
+
+        if (iq == null) {
+            debug(@"MAM: [%s] IQ null", account.bare_jid.to_string());
+            return true;
+        }
+
+        if (iq.stanza.get_deep_string_content("urn:xmpp:mam:2:fin", "http://jabber.org/protocol/rsm" + ":set", "first") == null) {
+            return true;
+        }
+
+        while (iq != null) {
+            debug("MAM: [%s] IN: %s", account.bare_jid.to_string(), iq.stanza.to_string());
+            string? earliest_id = iq.stanza.get_deep_string_content("urn:xmpp:mam:2:fin", "http://jabber.org/protocol/rsm" + ":set", "first");
+            if (earliest_id == null) return true;
+
+            if (!mam_times[account].has_key(earliest_id)) error("wtf");
+
+            debug("MAM: [%s] Update from_id %s\n", account.bare_jid.to_string(), earliest_id);
+            if (!current_catchup_id.has_key(account)) {
+                debug("MAM: [%s] We get our first MAM page", account.bare_jid.to_string());
+                string? latest_id = iq.stanza.get_deep_string_content("urn:xmpp:mam:2:fin", "http://jabber.org/protocol/rsm" + ":set", "last");
+                if (!mam_times[account].has_key(latest_id)) error("wtf2");
+                current_catchup_id[account] = (int) db.mam_catchup.insert()
+                        .value(db.mam_catchup.account_id, account.id)
+                        .value(db.mam_catchup.from_id, earliest_id)
+                        .value(db.mam_catchup.from_time, (long)mam_times[account][earliest_id].to_unix())
+                        .value(db.mam_catchup.to_id, latest_id)
+                        .value(db.mam_catchup.to_time, (long)mam_times[account][latest_id].to_unix())
+                        .perform();
+            } else {
+                // Update existing id
+                db.mam_catchup.update()
+                        .set(db.mam_catchup.from_id, earliest_id)
+                        .set(db.mam_catchup.from_time, (long)mam_times[account][earliest_id].to_unix()) // need to make sure we have this
+                        .with(db.mam_catchup.id, "=", current_catchup_id[account])
+                        .perform();
+            }
+
+            TimeSpan catchup_time_ago = (new DateTime.now_utc()).difference(mam_times[account][earliest_id]);
+            int wait_ms = 10;
+            if (catchup_time_ago > 14 * TimeSpan.DAY) {
+                wait_ms = 2000;
+            } else if (catchup_time_ago > 5 * TimeSpan.DAY) {
+                wait_ms = 1000;
+            } else if (catchup_time_ago > 2 * TimeSpan.DAY) {
+                wait_ms = 200;
+            } else if (catchup_time_ago > TimeSpan.DAY) {
+                wait_ms = 50;
+            }
+
+            mam_times[account] = new HashMap<string, DateTime>();
+
+            Timeout.add(wait_ms, () => {
+                if (hitted_range.has_key(query_id)) {
+                    debug(@"MAM: [%s] Hitted contains key %s", account.bare_jid.to_string(), query_id);
+                    iq = null;
+                    Idle.add(get_mam_range.callback);
+                    return false;
+                }
+
+                stream.get_module(Xep.MessageArchiveManagement.Module.IDENTITY).page_through_results.begin(stream, null, query_id, from_time, to_time, iq, (_, res) => {
+                    iq = stream.get_module(Xep.MessageArchiveManagement.Module.IDENTITY).page_through_results.end(res);
+                    Idle.add(get_mam_range.callback);
+                });
+                return false;
+            });
+            yield;
+        }
+        return false;
+    }
+
     private async void on_message_received(Account account, Xmpp.MessageStanza message_stanza) {
         Entities.Message message = yield parse_message_stanza(account, message_stanza);
 
         Conversation? conversation = stream_interactor.get_module(ConversationManager.IDENTITY).get_conversation_for_message(message);
-        if (conversation != null) {
-            bool abort = yield received_pipeline.run(message, message_stanza, conversation);
-            if (abort) return;
+        if (conversation == null) return;
+
+        // MAM state database update
+        Xep.MessageArchiveManagement.MessageFlag mam_flag = Xep.MessageArchiveManagement.MessageFlag.get_flag(message_stanza);
+        if (mam_flag == null) {
+            if (current_catchup_id.has_key(account)) {
+                string? stanza_id = UniqueStableStanzaIDs.get_stanza_id(message_stanza, account.bare_jid);
+                if (stanza_id != null) {
+                    db.mam_catchup.update()
+                        .with(db.mam_catchup.id, "=", current_catchup_id[account])
+                        .set(db.mam_catchup.to_time, (long)message.local_time.to_unix())
+                        .set(db.mam_catchup.to_id, stanza_id)
+                        .perform();
+                }
+            }
         }
+
+        bool abort = yield received_pipeline.run(message, message_stanza, conversation);
+        if (abort) return;
+
         if (message.direction == Entities.Message.DIRECTION_RECEIVED) {
             message_received(message, conversation);
         } else if (message.direction == Entities.Message.DIRECTION_SENT) {
@@ -170,24 +391,78 @@ public class MessageProcessor : StreamInteractionModule, Object {
         public override string action_group { get { return "DEDUPLICATE"; } }
         public override string[] after_actions { get { return after_actions_const; } }
 
+        private MessageProcessor outer;
         private Database db;
 
-        public DeduplicateMessageListener(Database db) {
+        public DeduplicateMessageListener(MessageProcessor outer, Database db) {
+            this.outer = outer;
             this.db = db;
         }
 
         public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
+            Account account = conversation.account;
+
+            Xep.MessageArchiveManagement.MessageFlag? mam_flag = Xep.MessageArchiveManagement.MessageFlag.get_flag(stanza);
+
+            // Deduplicate by server_id
             if (message.server_id != null) {
-                return db.contains_message_by_server_id(conversation.account, message.counterpart, message.server_id);
-            } else if (message.stanza_id != null) {
-                bool is_uuid = Regex.match_simple("""[0-9A-Fa-f]{8}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{12}""", message.stanza_id);
-                if (is_uuid) {
-                    return db.contains_message_by_stanza_id(message, conversation.account);
+                QueryBuilder builder =  db.message.select()
+                        .with(db.message.server_id, "=", message.server_id)
+                        .with(db.message.counterpart_id, "=", db.get_jid_id(message.counterpart))
+                        .with(db.message.account_id, "=", account.id);
+                bool duplicate = builder.count() > 0;
+
+                if (duplicate && mam_flag != null) {
+                    debug(@"MAM: [%s] Hitted range duplicate server id. id %s qid %s", account.bare_jid.to_string(), message.server_id, mam_flag.query_id);
+                    if (outer.catchup_until_time.has_key(account) && mam_flag.server_time.compare(outer.catchup_until_time[account]) < 0) {
+                        outer.hitted_range[mam_flag.query_id] = -1;
+                        debug(@"MAM: [%s] In range (time) %s < %s", account.bare_jid.to_string(), mam_flag.server_time.to_string(), outer.catchup_until_time[account].to_string());
+                    }
+                }
+                if (duplicate) return true;
+            }
+
+            // Deduplicate messages by uuid
+            bool is_uuid = message.stanza_id != null && Regex.match_simple("""[0-9A-Fa-f]{8}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{12}""", message.stanza_id);
+            if (is_uuid) {
+                QueryBuilder builder =  db.message.select()
+                        .with(db.message.stanza_id, "=", message.stanza_id)
+                        .with(db.message.counterpart_id, "=", db.get_jid_id(message.counterpart))
+                        .with(db.message.account_id, "=", account.id);
+                if (message.counterpart.resourcepart != null) {
+                    builder.with(db.message.counterpart_resource, "=", message.counterpart.resourcepart);
                 } else {
-                    return db.contains_message(message, conversation.account);
+                    builder.with_null(db.message.counterpart_resource);
                 }
+                RowOption row_opt = builder.single().row();
+                bool duplicate = row_opt.is_present();
+
+                if (duplicate && mam_flag != null && row_opt[db.message.server_id] == null &&
+                        outer.catchup_until_time.has_key(account) && mam_flag.server_time.compare(outer.catchup_until_time[account]) > 0) {
+                    outer.hitted_range[mam_flag.query_id] = -1;
+                    debug(@"MAM: [%s] Hitted range duplicate message id. id %s qid %s", account.bare_jid.to_string(), message.stanza_id, mam_flag.query_id);
+                }
+                return duplicate;
             }
-            return false;
+
+            // Deduplicate messages based on content and metadata
+            QueryBuilder builder = db.message.select()
+                    .with(db.message.account_id, "=", account.id)
+                    .with(db.message.counterpart_id, "=", db.get_jid_id(message.counterpart))
+                    .with(db.message.body, "=", message.body)
+                    .with(db.message.time, "<", (long) message.time.add_minutes(1).to_unix())
+                    .with(db.message.time, ">", (long) message.time.add_minutes(-1).to_unix());
+            if (message.stanza_id != null) {
+                builder.with(db.message.stanza_id, "=", message.stanza_id);
+            } else {
+                builder.with_null(db.message.stanza_id);
+            }
+            if (message.counterpart.resourcepart != null) {
+                builder.with(db.message.counterpart_resource, "=", message.counterpart.resourcepart);
+            } else {
+                builder.with_null(db.message.counterpart_resource);
+            }
+            return builder.count() > 0;
         }
     }
 
diff --git a/libdino/src/service/notification_events.vala b/libdino/src/service/notification_events.vala
index f47b9a0a..dca81af8 100644
--- a/libdino/src/service/notification_events.vala
+++ b/libdino/src/service/notification_events.vala
@@ -16,9 +16,6 @@ public class NotificationEvents : StreamInteractionModule, Object {
 
     private StreamInteractor stream_interactor;
 
-    private HashMap<Account, HashMap<Conversation, ContentItem>> mam_potential_new = new HashMap<Account, HashMap<Conversation, ContentItem>>(Account.hash_func, Account.equals_func);
-    private Gee.List<Account> synced_accounts = new ArrayList<Account>(Account.equals_func);
-
     public static void start(StreamInteractor stream_interactor) {
         NotificationEvents m = new NotificationEvents(stream_interactor);
         stream_interactor.add_module(m);
@@ -31,36 +28,19 @@ public class NotificationEvents : StreamInteractionModule, Object {
         stream_interactor.get_module(PresenceManager.IDENTITY).received_subscription_request.connect(on_received_subscription_request);
         stream_interactor.get_module(MucManager.IDENTITY).invite_received.connect((account, room_jid, from_jid, password, reason) => notify_muc_invite(account, room_jid, from_jid, password, reason));
         stream_interactor.connection_manager.connection_error.connect((account, error) => notify_connection_error(account, error));
-        stream_interactor.get_module(MessageProcessor.IDENTITY).history_synced.connect((account) => {
-            synced_accounts.add(account);
-            if (!mam_potential_new.has_key(account)) return;
-            foreach (Conversation c in mam_potential_new[account].keys) {
-                ContentItem last_mam_item = mam_potential_new[account][c];
-                ContentItem last_item = stream_interactor.get_module(ContentItemStore.IDENTITY).get_latest(c);
-                if (last_mam_item == last_item /* && !c.read_up_to.equals(m) */) {
-                    on_content_item_received(last_mam_item, c);
-                }
-            }
-            mam_potential_new[account].clear();
-        });
     }
 
     private void on_content_item_received(ContentItem item, Conversation conversation) {
 
-        // Don't wait for MAM sync on servers without MAM
-        bool mam_available = true;
-        XmppStream? stream = stream_interactor.get_stream(conversation.account);
-        if (stream != null) {
-            mam_available = stream.get_flag(Xep.MessageArchiveManagement.Flag.IDENTITY) != null;
-        }
+        ContentItem last_item = stream_interactor.get_module(ContentItemStore.IDENTITY).get_latest(conversation);
 
-        if (mam_available && !synced_accounts.contains(conversation.account)) {
-            if (!mam_potential_new.has_key(conversation.account)) {
-                mam_potential_new[conversation.account] = new HashMap<Conversation, ContentItem>(Conversation.hash_func, Conversation.equals_func);
-            }
-            mam_potential_new[conversation.account][conversation] = item;
-            return;
+        bool not_read_up_to = true;
+        MessageItem message_item = item as MessageItem;
+        if (message_item != null) {
+            not_read_up_to = conversation.read_up_to != null && !conversation.read_up_to.equals(message_item.message);
         }
+        if (item.id != last_item.id && not_read_up_to) return;
+
         if (!should_notify(item, conversation)) return;
         if (stream_interactor.get_module(ChatInteraction.IDENTITY).is_active_focus()) return;
         notify_content_item(item, conversation);
diff --git a/plugins/crypto-vala/src/cipher_converter.vala b/plugins/crypto-vala/src/cipher_converter.vala
index f1363fb0..b2b52c5a 100644
--- a/plugins/crypto-vala/src/cipher_converter.vala
+++ b/plugins/crypto-vala/src/cipher_converter.vala
@@ -85,9 +85,7 @@ public class SymmetricCipherDecrypter : SymmetricCipherConverter {
             inbuf.length += (int) attached_taglen;
             if ((flags & ConverterFlags.INPUT_AT_END) != 0) {
                 if (attached_taglen > 0) {
-                    print("Checking tag\n");
                     check_tag(inbuf[(inbuf.length - attached_taglen):inbuf.length]);
-                    print("tag ok\n");
                     bytes_read = inbuf.length;
                 }
                 return ConverterResult.FINISHED;
@@ -101,4 +99,4 @@ public class SymmetricCipherDecrypter : SymmetricCipherConverter {
         }
     }
 }
-}
\ No newline at end of file
+}
diff --git a/xmpp-vala/src/module/message/module.vala b/xmpp-vala/src/module/message/module.vala
index ab3a7d80..5ddbbe1a 100644
--- a/xmpp-vala/src/module/message/module.vala
+++ b/xmpp-vala/src/module/message/module.vala
@@ -12,6 +12,7 @@ namespace Xmpp {
         public StanzaListenerHolder<MessageStanza> send_pipeline = new StanzaListenerHolder<MessageStanza>();
 
         public signal void received_message(XmppStream stream, MessageStanza message);
+        public signal void received_message_unprocessed(XmppStream stream, MessageStanza message);
 
         public void send_message(XmppStream stream, MessageStanza message) {
             send_pipeline.run.begin(stream, message, (obj, res) => {
@@ -21,6 +22,9 @@ namespace Xmpp {
 
         public async void received_message_stanza_async(XmppStream stream, StanzaNode node) {
             MessageStanza message = new MessageStanza.from_stanza(node, stream.get_flag(Bind.Flag.IDENTITY).my_jid);
+
+            received_message_unprocessed(stream, message);
+
             if (!message.is_error()) {
                 bool abort = yield received_pipeline.run(stream, message);
                 if (abort) return;
diff --git a/xmpp-vala/src/module/xep/0045_muc/module.vala b/xmpp-vala/src/module/xep/0045_muc/module.vala
index b30145ff..59d61f3d 100644
--- a/xmpp-vala/src/module/xep/0045_muc/module.vala
+++ b/xmpp-vala/src/module/xep/0045_muc/module.vala
@@ -487,16 +487,15 @@ public class ReceivedPipelineListener : StanzaListener<MessageStanza> {
                 StanzaNode? invite_node = x_node.get_subnode("invite", NS_URI_USER);
                 string? password = null;
                 StanzaNode? password_node = x_node.get_subnode("password", NS_URI_USER);
-                if (password_node != null)
-                password = password_node.get_string_content();
+                if (password_node != null) password = password_node.get_string_content();
                 if (invite_node != null) {
                     string? from_jid = invite_node.get_attribute("from");
                     if (from_jid != null) {
                         StanzaNode? reason_node = invite_node.get_subnode("reason", NS_URI_USER);
                         string? reason = null;
-                        if (reason_node != null)
-                        reason = reason_node.get_string_content();
-                        outer.invite_received(stream, message.from, new Jid(from_jid), password, reason);
+                        if (reason_node != null) reason = reason_node.get_string_content();
+                        bool is_mam_message = Xep.MessageArchiveManagement.MessageFlag.get_flag(message) != null; // TODO
+                        if (!is_mam_message) outer.invite_received(stream, message.from, new Jid(from_jid), password, reason);
                         return true;
                     }
                 }
diff --git a/xmpp-vala/src/module/xep/0313_message_archive_management.vala b/xmpp-vala/src/module/xep/0313_message_archive_management.vala
index 4f8cadec..247e1b8e 100644
--- a/xmpp-vala/src/module/xep/0313_message_archive_management.vala
+++ b/xmpp-vala/src/module/xep/0313_message_archive_management.vala
@@ -14,10 +14,7 @@ public class Module : XmppStreamModule {
 
     private ReceivedPipelineListener received_pipeline_listener = new ReceivedPipelineListener();
 
-    public delegate void OnFinished(XmppStream stream);
-    public void query_archive(XmppStream stream, string? jid, DateTime? start, DateTime? end, owned OnFinished? on_finished = null) {
-        if (stream.get_flag(Flag.IDENTITY) == null) return;
-
+    private StanzaNode crate_base_query(XmppStream stream, string? jid, string? queryid, DateTime? start, DateTime? end) {
         DataForms.DataForm data_form = new DataForms.DataForm();
         DataForms.DataForm.HiddenField form_type_field = new DataForms.DataForm.HiddenField() { var="FORM_TYPE" };
         form_type_field.set_value_string(NS_VER(stream));
@@ -38,8 +35,41 @@ public class Module : XmppStreamModule {
             data_form.add_field(field);
         }
         StanzaNode query_node = new StanzaNode.build("query", NS_VER(stream)).add_self_xmlns().put_node(data_form.get_submit_node());
+        if (queryid != null) {
+            query_node.put_attribute("queryid", queryid);
+        }
+        return query_node;
+    }
+
+    private StanzaNode create_set_rsm_node(string? before_id) {
+        var before_node = new StanzaNode.build("before", "http://jabber.org/protocol/rsm");
+        if (before_id != null) {
+            before_node.put_node(new StanzaNode.text(before_id));
+        }
+        var max_node = (new StanzaNode.build("max", "http://jabber.org/protocol/rsm")).put_node(new StanzaNode.text("20"));
+        return (new StanzaNode.build("set", "http://jabber.org/protocol/rsm")).add_self_xmlns()
+                .put_node(before_node)
+                .put_node(max_node);
+    }
+
+    public async Iq.Stanza? query_archive(XmppStream stream, string? jid, string? query_id, DateTime? start_time, string? start_id, DateTime? end_time, string? end_id) {
+        if (stream.get_flag(Flag.IDENTITY) == null) return null;
+
+        var query_node = crate_base_query(stream, jid, query_id, start_time, end_time);
+
+        query_node.put_node(create_set_rsm_node(end_id));
         Iq.Stanza iq = new Iq.Stanza.set(query_node);
-        stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq, (stream, iq) => { page_through_results(stream, iq, (owned)on_finished); });
+
+        debug(@"OUT INIT: %s", iq.stanza.to_string());
+
+        Iq.Stanza? result_iq = null;
+        stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq, (stream, iq) => {
+            result_iq = iq;
+            Idle.add(query_archive.callback);
+        });
+        yield;
+
+        return result_iq;
     }
 
     public override void attach(XmppStream stream) {
@@ -54,22 +84,30 @@ public class Module : XmppStreamModule {
     public override string get_ns() { return NS_URI; }
     public override string get_id() { return IDENTITY.id; }
 
-    private static void page_through_results(XmppStream stream, Iq.Stanza iq, owned OnFinished? on_finished = null) {
-        string? last = iq.stanza.get_deep_string_content(NS_VER(stream) + ":fin", "http://jabber.org/protocol/rsm" + ":set", "last");
-        if (last == null) {
-            stream.get_flag(Flag.IDENTITY).cought_up = true;
-            if (on_finished != null) on_finished(stream);
-            return;
+    public async Iq.Stanza? page_through_results(XmppStream stream, string? jid, string? query_id, DateTime? start_time, DateTime? end_time, Iq.Stanza iq) {
+
+        string? complete = iq.stanza.get_deep_attribute("urn:xmpp:mam:2:fin", "complete");
+        if (complete == "true") {
+            return null;
+        }
+        string? first = iq.stanza.get_deep_string_content(NS_VER(stream) + ":fin", "http://jabber.org/protocol/rsm" + ":set", "first");
+        if (first == null) {
+            return null;
         }
 
-        Iq.Stanza paging_iq = new Iq.Stanza.set(
-                new StanzaNode.build("query", NS_VER(stream)).add_self_xmlns().put_node(
-                    new StanzaNode.build("set", "http://jabber.org/protocol/rsm").add_self_xmlns().put_node(
-                        new StanzaNode.build("after", "http://jabber.org/protocol/rsm").put_node(new StanzaNode.text(last))
-                    )
-                )
-            );
-        stream.get_module(Iq.Module.IDENTITY).send_iq(stream, paging_iq, (stream, iq) => { page_through_results(stream, iq, (owned)on_finished); });
+        var query_node = crate_base_query(stream, jid, query_id, start_time, end_time);
+        query_node.put_node(create_set_rsm_node(first));
+
+        Iq.Stanza paging_iq = new Iq.Stanza.set(query_node);
+
+        Iq.Stanza? result_iq = null;
+        stream.get_module(Iq.Module.IDENTITY).send_iq(stream, paging_iq, (stream, iq) => {
+            result_iq = iq;
+            Idle.add(page_through_results.callback);
+        });
+        yield;
+
+        return result_iq;
     }
 
     private void query_availability(XmppStream stream) {
@@ -107,7 +145,8 @@ public class ReceivedPipelineListener : StanzaListener<MessageStanza> {
             StanzaNode? forward_node = message.stanza.get_deep_subnode(NS_VER(stream) + ":result", "urn:xmpp:forward:0:forwarded", DelayedDelivery.NS_URI + ":delay");
             DateTime? datetime = DelayedDelivery.Module.get_time_for_node(forward_node);
             string? mam_id = message.stanza.get_deep_attribute(NS_VER(stream) + ":result", NS_VER(stream) + ":id");
-            message.add_flag(new MessageFlag(datetime, mam_id));
+            string? query_id = message.stanza.get_deep_attribute(NS_VER(stream) + ":result", NS_VER(stream) + ":queryid");
+            message.add_flag(new MessageFlag(datetime, mam_id, query_id));
 
             message.stanza = message_node;
             message.rerun_parsing = true;
@@ -134,10 +173,12 @@ public class MessageFlag : Xmpp.MessageFlag {
 
     public DateTime? server_time { get; private set; }
     public string? mam_id { get; private set; }
+    public string? query_id { get; private set; }
 
-    public MessageFlag(DateTime? server_time, string? mam_id) {
+    public MessageFlag(DateTime? server_time, string? mam_id, string? query_id) {
         this.server_time = server_time;
         this.mam_id = mam_id;
+        this.query_id = query_id;
     }
 
     public static MessageFlag? get_flag(MessageStanza message) { return (MessageFlag) message.get_flag(NS_URI, ID); }