Add support for XEP-0490 Message Displayed Synchronization

6 files changed, 167 insertions, 74 deletions

diff --git a/libdino/src/service/chat_interaction.vala b/libdino/src/service/chat_interaction.vala
index 1254a574..9c8106a0 100644
--- a/libdino/src/service/chat_interaction.vala
+++ b/libdino/src/service/chat_interaction.vala
@@ -18,6 +18,7 @@ public class ChatInteraction : StreamInteractionModule, Object {
     private HashMap<Conversation, DateTime> last_input_interaction = new HashMap<Conversation, DateTime>(Conversation.hash_func, Conversation.equals_func);
     private HashMap<Conversation, DateTime> last_interface_interaction = new HashMap<Conversation, DateTime>(Conversation.hash_func, Conversation.equals_func);
     private bool focus_in = false;
+    private bool scrolled_down = true;
 
     public static void start(StreamInteractor stream_interactor) {
         ChatInteraction m = new ChatInteraction(stream_interactor);
@@ -66,6 +67,14 @@ public class ChatInteraction : StreamInteractionModule, Object {
         on_conversation_unfocused(conversation);
     }
 
+    public void on_scrolled_down_changed(bool scrolled_down, Conversation? conversation) {
+        if (scrolled_down == this.scrolled_down) return;
+        this.scrolled_down = scrolled_down;
+        if (this.scrolled_down && focus_in && selected_conversation == conversation) {
+            mark_latest_message_displayed();
+        }
+    }
+
     public void on_message_entered(Conversation? conversation) {
         if (!last_input_interaction.has_key(conversation)) {
             send_chat_state_notification(conversation, Xep.ChatStateNotifications.STATE_COMPOSING);
@@ -84,25 +93,22 @@ public class ChatInteraction : StreamInteractionModule, Object {
     public void on_conversation_selected(Conversation conversation) {
         on_conversation_unfocused(selected_conversation);
         selected_conversation = conversation;
+        scrolled_down = false;
         on_conversation_focused(conversation);
     }
 
     private void new_item(ContentItem item, Conversation conversation) {
-        bool mark_read = is_active_focus(conversation);
+        bool mark_read = is_active_focus(conversation) && scrolled_down;
 
         if (!mark_read) {
             MessageItem? message_item = item as MessageItem;
-            if (message_item != null) {
-                if (message_item.message.direction == Message.DIRECTION_SENT) {
-                    mark_read = true;
-                }
+            if (message_item != null && message_item.message.direction == Message.DIRECTION_SENT) {
+                mark_read = true;
             }
             if (message_item == null) {
                 FileItem? file_item = item as FileItem;
-                if (file_item != null) {
-                    if (file_item.file_transfer.direction == FileTransfer.DIRECTION_SENT) {
-                        mark_read = true;
-                    }
+                if (file_item != null && file_item.file_transfer.direction == FileTransfer.DIRECTION_SENT) {
+                    mark_read = true;
                 }
             }
         }
@@ -127,12 +133,8 @@ public class ChatInteraction : StreamInteractionModule, Object {
         focus_in = true;
         if (conversation == null) return;
         focused_in(conversation);
-        check_send_read();
 
-        ContentItem? latest_item = stream_interactor.get_module(ContentItemStore.IDENTITY).get_latest(conversation);
-        if (latest_item != null) {
-            conversation.read_up_to_item = latest_item.id;
-        }
+        if (scrolled_down) mark_latest_message_displayed();
     }
 
     private void on_conversation_unfocused(Conversation? conversation) {
@@ -145,12 +147,15 @@ public class ChatInteraction : StreamInteractionModule, Object {
         }
     }
 
-    private void check_send_read() {
-        if (selected_conversation == null) return;
-        Entities.Message? message = stream_interactor.get_module(MessageStorage.IDENTITY).get_last_message(selected_conversation);
-        if (message != null && message.direction == Entities.Message.DIRECTION_RECEIVED) {
-            send_chat_marker(message, null, selected_conversation, Xep.ChatMarkers.MARKER_DISPLAYED);
-        }
+    private async bool mark_latest_message_displayed() {
+        Conversation conversation = selected_conversation;
+        if (conversation == null) return false;
+        Entities.Message? message = stream_interactor.get_module(MessageStorage.IDENTITY).get_last_message(conversation);
+        if (message == null || message.direction == Entities.Message.DIRECTION_SENT) return false;
+        if (message.equals(conversation.read_up_to)) return false;
+
+        bool sent_marker = yield send_and_mark_message_displayed(conversation, message);
+        return sent_marker;
     }
 
     private bool update_interactions() {
@@ -195,43 +200,67 @@ public class ChatInteraction : StreamInteractionModule, Object {
 
             // Send chat marker
             if (message.direction == Entities.Message.DIRECTION_SENT) return false;
-            if (outer.is_active_focus(conversation)) {
-                outer.check_send_read();
-                outer.send_chat_marker(message, stanza, conversation, Xep.ChatMarkers.MARKER_DISPLAYED);
+            if (outer.is_active_focus(conversation) && outer.scrolled_down) {
+                yield outer.send_and_mark_message_displayed(conversation, message);
             } else {
-                outer.send_chat_marker(message, stanza, conversation, Xep.ChatMarkers.MARKER_RECEIVED);
+                outer.send_received_marker(message, stanza, conversation);
             }
             return false;
         }
     }
 
+    public async bool send_and_mark_message_displayed(Conversation conversation, Entities.Message message) {
+        stream_interactor.get_module(CounterpartInteractionManager.IDENTITY).mark_displayed_up_to_message(conversation, message);
+        bool sent_marker = send_displayed_marker(message, conversation);
+        if (!sent_marker) {
+            sent_marker = yield send_message_displayed_sync(message, conversation);
+        }
+        return sent_marker;
+    }
+
+    private bool send_received_marker(Entities.Message message, Xmpp.MessageStanza? stanza, Conversation conversation) {
+        if (stanza == null) return false;
+        if (!Xep.ChatMarkers.Module.requests_marking(stanza)) return false;
+        if (message.type_ == Message.Type.GROUPCHAT) return false;
+        if (message.stanza_id == null) return false;
+        if (message.direction == Entities.Message.DIRECTION_SENT) return false;
+        if (message.equals(conversation.read_up_to)) return false;
+        if (conversation.get_send_marker_setting(stream_interactor) != Conversation.Setting.ON) return false;
 
-    private void send_chat_marker(Entities.Message message, Xmpp.MessageStanza? stanza, Conversation conversation, string marker) {
         XmppStream? stream = stream_interactor.get_stream(conversation.account);
-        if (stream == null) return;
+        if (stream == null) return false;
 
-        switch (marker) {
-            case Xep.ChatMarkers.MARKER_RECEIVED:
-                if (stanza != null && Xep.ChatMarkers.Module.requests_marking(stanza) && message.type_ != Message.Type.GROUPCHAT) {
-                    if (message.stanza_id == null) return;
-                    stream.get_module(Xep.ChatMarkers.Module.IDENTITY).send_marker(stream, message.from, message.stanza_id, message.get_type_string(), Xep.ChatMarkers.MARKER_RECEIVED);
-                }
-                break;
-            case Xep.ChatMarkers.MARKER_DISPLAYED:
-                if (conversation.get_send_marker_setting(stream_interactor) == Conversation.Setting.ON) {
-                    if (message.equals(conversation.read_up_to)) return;
-                    conversation.read_up_to = message;
+        stream.get_module(Xep.ChatMarkers.Module.IDENTITY).send_marker(stream, message.from, message.stanza_id, message.get_type_string(), Xep.ChatMarkers.MARKER_RECEIVED);
+        return true;
+    }
 
-                    if (message.type_ == Message.Type.GROUPCHAT || message.type_ == Message.Type.GROUPCHAT_PM) {
-                        if (message.server_id == null) return;
-                        stream.get_module(Xep.ChatMarkers.Module.IDENTITY).send_marker(stream, message.from.bare_jid, message.server_id, message.get_type_string(), Xep.ChatMarkers.MARKER_DISPLAYED);
-                    } else {
-                        if (message.stanza_id == null) return;
-                        stream.get_module(Xep.ChatMarkers.Module.IDENTITY).send_marker(stream, message.from, message.stanza_id, message.get_type_string(), Xep.ChatMarkers.MARKER_DISPLAYED);
-                    }
-                }
-                break;
+    private bool send_displayed_marker(Entities.Message message, Conversation conversation) {
+        if (message.direction == Entities.Message.DIRECTION_SENT) return false;
+        if (conversation.get_send_marker_setting(stream_interactor) != Conversation.Setting.ON) return false;
+
+        XmppStream? stream = stream_interactor.get_stream(conversation.account);
+        if (stream == null) return false;
+
+        if (message.type_.is_muc_semantic() && message.server_id != null) {
+            stream.get_module(Xep.ChatMarkers.Module.IDENTITY).send_marker(stream, message.from.bare_jid, message.server_id, message.get_type_string(), Xep.ChatMarkers.MARKER_DISPLAYED);
+            return true;
+        } else if (!message.type_.is_muc_semantic() && message.stanza_id != null) {
+            stream.get_module(Xep.ChatMarkers.Module.IDENTITY).send_marker(stream, message.from, message.stanza_id, message.get_type_string(), Xep.ChatMarkers.MARKER_DISPLAYED);
+            return true;
         }
+
+        return false;
+    }
+
+    private async bool send_message_displayed_sync(Entities.Message message, Conversation conversation) {
+        if (message.direction == Entities.Message.DIRECTION_SENT) return false;
+        if (message.server_id == null) return false;
+
+        XmppStream? stream = stream_interactor.get_stream(conversation.account);
+        if (stream == null) return false;
+
+        Jid server_id_jid = conversation.type_ == Conversation.Type.GROUPCHAT ? conversation.counterpart : conversation.account.bare_jid;
+        return yield stream.get_module(Xep.MessageDisplayedSynchronization.Module.IDENTITY).update_message_displayed(stream, conversation.counterpart, server_id_jid, message.server_id);
     }
 
     private void send_delivery_receipt(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
diff --git a/libdino/src/service/content_item_store.vala b/libdino/src/service/content_item_store.vala
index 37c82eb7..d74f9b05 100644
--- a/libdino/src/service/content_item_store.vala
+++ b/libdino/src/service/content_item_store.vala
@@ -76,10 +76,7 @@ public class ContentItemStore : StreamInteractionModule, Object {
             case 2:
                 FileTransfer? file_transfer = stream_interactor.get_module(FileTransferStorage.IDENTITY).get_file_by_id(foreign_id, conversation);
                 if (file_transfer != null) {
-                    Message? message = null;
-                    if (file_transfer.provider == 0 && file_transfer.info != null) {
-                        message = stream_interactor.get_module(MessageStorage.IDENTITY).get_message_by_id(int.parse(file_transfer.info), conversation);
-                    }
+                    Message? message = stream_interactor.get_module(FileManager.IDENTITY).get_message_for_file_transfer(file_transfer, conversation);
                     var file_item = new FileItem(file_transfer, conversation, id, message);
                     return file_item;
                 }
diff --git a/libdino/src/service/counterpart_interaction_manager.vala b/libdino/src/service/counterpart_interaction_manager.vala
index 23db5762..f60cfb20 100644
--- a/libdino/src/service/counterpart_interaction_manager.vala
+++ b/libdino/src/service/counterpart_interaction_manager.vala
@@ -73,6 +73,9 @@ public class CounterpartInteractionManager : StreamInteractionModule, Object {
         stream_interactor.module_manager.get_module(account, Xep.ChatStateNotifications.Module.IDENTITY).chat_state_received.connect((stream, jid, state, stanza) => {
             on_chat_state_received.begin(account, jid, state, stanza);
         });
+        stream_interactor.module_manager.get_module(account, Xep.MessageDisplayedSynchronization.Module.IDENTITY).on_message_displayed.connect((conversation_jid, stanza_id_by_jid, stanza_id) => {
+            on_message_displayed_received(account, conversation_jid, stanza_id_by_jid, stanza_id);
+        });
     }
 
     private void clear_chat_state(Conversation conversation, Jid jid) {
@@ -120,17 +123,69 @@ public class CounterpartInteractionManager : StreamInteractionModule, Object {
     private async void on_chat_marker_received(Account account, Jid jid, string marker, string stanza_id, MessageStanza message_stanza) {
         Conversation? conversation = stream_interactor.get_module(ConversationManager.IDENTITY).approx_conversation_for_stanza(message_stanza.from, message_stanza.to, account, message_stanza.type_);
         if (conversation == null) return;
-        handle_chat_marker(conversation, jid, marker, stanza_id);
+        handle_chat_marker(conversation, jid, marker, stanza_id, message_stanza);
+    }
+
+    private void on_message_displayed_received(Account account, Jid conversation_jid, Jid stanza_id_by_jid, string stanza_id) {
+        Conversation? conversation = null;
+        if (stanza_id_by_jid.equals(conversation_jid) && conversation_jid.is_bare()) {
+            conversation = stream_interactor.get_module(ConversationManager.IDENTITY).get_conversation(conversation_jid, account, Conversation.Type.GROUPCHAT);
+        } else if (conversation_jid.resourcepart != null && stanza_id_by_jid.equals(account.bare_jid)) {
+            conversation = stream_interactor.get_module(ConversationManager.IDENTITY).get_conversation(conversation_jid, account, Conversation.Type.GROUPCHAT_PM);
+        } else if (stanza_id_by_jid.equals(account.bare_jid)) {
+            conversation = stream_interactor.get_module(ConversationManager.IDENTITY).get_conversation(conversation_jid, account, Conversation.Type.CHAT);
+        }
+        if (conversation == null) return;
+
+        mark_displayed_up_to_stanza_id(conversation, stanza_id, true);
+    }
+
+    public bool mark_displayed_up_to_stanza_id(Conversation conversation, string stanza_id, bool server_id_only) {
+        Entities.Message? message = null;
+        if (server_id_only) {
+            message = stream_interactor.get_module(MessageStorage.IDENTITY).get_message_by_server_id(stanza_id, conversation);
+        } else if (conversation.type_ == Conversation.Type.GROUPCHAT || conversation.type_ == Conversation.Type.GROUPCHAT_PM) {
+            message = stream_interactor.get_module(MessageStorage.IDENTITY).get_message_by_server_id(stanza_id, conversation);
+            // Outdated clients might use the message id. Or in MUCs that don't send server ids.
+            if (message == null) {
+                message = stream_interactor.get_module(MessageStorage.IDENTITY).get_message_by_stanza_id(stanza_id, conversation);
+            }
+        } else {
+            message = stream_interactor.get_module(MessageStorage.IDENTITY).get_message_by_stanza_id(stanza_id, conversation);
+        }
+        if (message == null) return false;
+        return mark_displayed_up_to_message(conversation, message);
+    }
+
+    public bool mark_displayed_up_to_message(Conversation conversation, Message message) {
+        // Don't move read marker backwards because we get old info from another client
+        if (conversation.read_up_to != null && conversation.read_up_to.local_time.compare(message.local_time) > 0) return false;
+        conversation.read_up_to = message;
+
+        ContentItem? content_item = stream_interactor.get_module(ContentItemStore.IDENTITY).get_content_item_for_message(conversation, message);
+        if (content_item == null) return true;
+        return mark_displayed_up_to_content_item(conversation, content_item);
+    }
+
+    public bool mark_displayed_up_to_content_item(Conversation conversation, ContentItem content_item) {
+        ContentItem? read_up_to_item = stream_interactor.get_module(ContentItemStore.IDENTITY).get_item_by_id(conversation, conversation.read_up_to_item);
+        if (read_up_to_item != null && read_up_to_item.compare(content_item) > 0) return false;
+        conversation.read_up_to_item = content_item.id;
+        return true;
     }
 
-    private void handle_chat_marker(Conversation conversation, Jid jid, string marker, string stanza_id) {
+    private void handle_chat_marker(Conversation conversation, Jid jid, string marker, string stanza_id, MessageStanza? message_stanza = null) {
         // Check if the marker comes from ourselves (own jid or our jid in a MUC)
         bool own_marker = false;
         if (conversation.type_ == Conversation.Type.CHAT) {
             own_marker = conversation.account.bare_jid.to_string() == jid.bare_jid.to_string();
         } else {
             Jid? own_muc_jid = stream_interactor.get_module(MucManager.IDENTITY).get_own_jid(jid.bare_jid, conversation.account);
+            string own_occupant_id = stream_interactor.get_module(MucManager.IDENTITY).get_own_occupant_id(conversation.account, jid.bare_jid);
+            string occupant_id = Xep.OccupantIds.get_occupant_id(message_stanza.stanza);
             if (own_muc_jid != null && own_muc_jid.equals(jid)) {
+                // Don't allow groupchat displayed marker without matching occupant id
+                if (conversation.type_ == Conversation.Type.GROUPCHAT && (occupant_id == null || own_occupant_id == occupant_id)) return;
                 own_marker = true;
             }
         }
@@ -138,27 +193,7 @@ public class CounterpartInteractionManager : StreamInteractionModule, Object {
         if (own_marker) {
             // If we received a display marker from ourselves (other device), set the conversation read up to that message.
             if (marker != Xep.ChatMarkers.MARKER_DISPLAYED && marker != Xep.ChatMarkers.MARKER_ACKNOWLEDGED) return;
-            Entities.Message? message = null;
-            if (conversation.type_ == Conversation.Type.GROUPCHAT || conversation.type_ == Conversation.Type.GROUPCHAT_PM) {
-                message = stream_interactor.get_module(MessageStorage.IDENTITY).get_message_by_server_id(stanza_id, conversation);
-                // Outdated clients might use the message id. Or in MUCs that don't send server ids.
-                if (message == null) {
-                    message = stream_interactor.get_module(MessageStorage.IDENTITY).get_message_by_stanza_id(stanza_id, conversation);
-                }
-            } else {
-                message = stream_interactor.get_module(MessageStorage.IDENTITY).get_message_by_stanza_id(stanza_id, conversation);
-            }
-            if (message == null) return;
-            // Don't move read marker backwards because we get old info from another client
-            if (conversation.read_up_to != null && conversation.read_up_to.local_time.compare(message.local_time) > 0) return;
-            conversation.read_up_to = message;
-
-            // TODO: This only marks messages as read, not http file transfers.
-            ContentItem? content_item = stream_interactor.get_module(ContentItemStore.IDENTITY).get_item_by_foreign(conversation, 1, message.id);
-            if (content_item == null) return;
-            ContentItem? read_up_to_item = stream_interactor.get_module(ContentItemStore.IDENTITY).get_item_by_id(conversation, conversation.read_up_to_item);
-            if (read_up_to_item != null && read_up_to_item.compare(content_item) > 0) return;
-            conversation.read_up_to_item = content_item.id;
+            mark_displayed_up_to_stanza_id(conversation, stanza_id, false);
         } else {
             // We can't currently handle chat markers in MUCs
             if (conversation.type_ == Conversation.Type.GROUPCHAT) return;
@@ -203,7 +238,7 @@ public class CounterpartInteractionManager : StreamInteractionModule, Object {
     private void on_receipt_received(Account account, Jid jid, string id, MessageStanza stanza) {
         Conversation? conversation = stream_interactor.get_module(ConversationManager.IDENTITY).approx_conversation_for_stanza(stanza.from, stanza.to, account, stanza.type_);
         if (conversation == null) return;
-        handle_chat_marker(conversation, jid,Xep.ChatMarkers.MARKER_RECEIVED, id);
+        handle_chat_marker(conversation, jid,Xep.ChatMarkers.MARKER_RECEIVED, id, stanza);
     }
 }
 
diff --git a/libdino/src/service/file_manager.vala b/libdino/src/service/file_manager.vala
index 8c196dd7..5e59aee1 100644
--- a/libdino/src/service/file_manager.vala
+++ b/libdino/src/service/file_manager.vala
@@ -420,6 +420,13 @@ public class FileManager : StreamInteractionModule, Object {
             throw new FileSendError.SAVE_FAILED("Saving file error: %s".printf(e.message));
         }
     }
+
+    public Message? get_message_for_file_transfer(FileTransfer file_transfer, Conversation conversation) {
+        if ((file_transfer.provider == 0 || file_transfer.provider == 2) && file_transfer.info != null) {
+            return stream_interactor.get_module(MessageStorage.IDENTITY).get_message_by_id(int.parse(file_transfer.info), conversation);
+        }
+        return null;
+    }
 }
 
 public errordomain FileSendError {
diff --git a/libdino/src/service/message_processor.vala b/libdino/src/service/message_processor.vala
index 676daa72..7201085e 100644
--- a/libdino/src/service/message_processor.vala
+++ b/libdino/src/service/message_processor.vala
@@ -39,6 +39,7 @@ public class MessageProcessor : StreamInteractionModule, Object {
         received_pipeline.connect(new StoreMessageListener(this, stream_interactor));
         received_pipeline.connect(new StoreContentItemListener(stream_interactor));
         received_pipeline.connect(new MarkupListener(stream_interactor));
+        received_pipeline.connect(new PreDisplayedHandler(stream_interactor));
 
         stream_interactor.account_added.connect(on_account_added);
 
@@ -379,6 +380,29 @@ public class MessageProcessor : StreamInteractionModule, Object {
         }
     }
 
+    private class PreDisplayedHandler : MessageListener {
+        public string[] after_actions_const = new string[]{ "STORE_CONTENT_ITEM" };
+        public override string action_group { get { return "DISPLAYED_HANDLING"; } }
+        public override string[] after_actions { get { return after_actions_const; } }
+
+        private StreamInteractor stream_interactor;
+
+        public PreDisplayedHandler(StreamInteractor stream_interactor) {
+            this.stream_interactor = stream_interactor;
+        }
+
+        public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
+            var flag = stream_interactor.get_stream(conversation.account).get_flag(MessageDisplayedSynchronization.Flag.IDENTITY);
+            if (flag != null && message.server_id != null) {
+                Jid stanza_id_by = message.type_ == Message.Type.GROUPCHAT ? conversation.counterpart.bare_jid : conversation.account.bare_jid;
+                if (flag.is_displayed(conversation.counterpart, stanza_id_by, message.server_id)) {
+                    stream_interactor.get_module(CounterpartInteractionManager.IDENTITY).mark_displayed_up_to_message(conversation, message);
+                }
+            }
+            return false;
+        }
+    }
+
     public Entities.Message create_out_message(string? text, Conversation conversation) {
         Entities.Message message = new Entities.Message(text);
         message.type_ = Util.get_message_type_for_conversation(conversation);
diff --git a/libdino/src/service/module_manager.vala b/libdino/src/service/module_manager.vala
index 127f2d93..f83ce425 100644
--- a/libdino/src/service/module_manager.vala
+++ b/libdino/src/service/module_manager.vala
@@ -100,6 +100,7 @@ public class ModuleManager {
             module_map[account].add(new Xep.Muji.Module());
             module_map[account].add(new Xep.CallInvites.Module());
             module_map[account].add(new Xep.Coin.Module());
+            module_map[account].add(new Xep.MessageDisplayedSynchronization.Module());
             initialize_account_modules(account, module_map[account]);
         }
     }