From c3532bdf3141bcf0cbf9e4ae7a926dcda4f132ef Mon Sep 17 00:00:00 2001
From: fiaxh <git@lightrise.org>
Date: Wed, 18 Dec 2019 18:53:14 +0100
Subject: Refactor MAM catchup. Fetch from latest to earliest message.
---
libdino/src/service/conversation_manager.vala | 2 +-
libdino/src/service/database.vala | 71 +++---
libdino/src/service/message_processor.vala | 313 ++++++++++++++++++++++++--
libdino/src/service/notification_events.vala | 34 +--
4 files changed, 330 insertions(+), 90 deletions(-)
(limited to 'libdino')
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);
--
cgit v1.2.3-70-g09d2