Matrix: re-add sync from latest cache

2 files changed, 213 insertions, 214 deletions

diff --git a/src/plugins/Matrix.cpp b/src/plugins/Matrix.cpp
index 9bbf9bd..63027ab 100644
--- a/src/plugins/Matrix.cpp
+++ b/src/plugins/Matrix.cpp
@@ -17,6 +17,7 @@
 #include <rapidjson/filewritestream.h>
 #include <fcntl.h>
 #include <unistd.h>
+#include <fstream>
 #include <malloc.h>
 #include "../../include/QuickMedia.hpp"
 #include <HtmlParser.h>
@@ -38,7 +39,6 @@ namespace QuickMedia {
     // then we cant see room message preview. TODO: Fix this somehow.
     // TODO: What about state events in initial sync in timeline? such as user display name change.
     static const char* INITIAL_FILTER = "{\"presence\":{\"limit\":0,\"types\":[\"\"]},\"account_data\":{\"types\":[\"qm.emoji\",\"m.direct\"]},\"room\":{\"state\":{\"not_types\":[\"m.room.related_groups\",\"m.room.power_levels\",\"m.room.join_rules\",\"m.room.history_visibility\",\"m.room.canonical_alias\",\"m.space.child\"],\"lazy_load_members\":true},\"timeline\":{\"types\":[\"m.room.message\"],\"limit\":1,\"lazy_load_members\":true},\"ephemeral\":{\"limit\":0,\"types\":[\"\"],\"lazy_load_members\":true},\"account_data\":{\"limit\":1,\"types\":[\"m.fully_read\",\"m.tag\",\"qm.last_read_message_timestamp\"],\"lazy_load_members\":true}}}";
-    static const char* ADDITIONAL_MESSAGES_FILTER = "{\"presence\":{\"types\":[\"\"]},\"account_data\":{\"limit\":0,\"types\":[\"\"]},\"room\":{\"state\":{\"not_types\":[\"m.room.related_groups\",\"m.room.power_levels\",\"m.room.join_rules\",\"m.room.history_visibility\",\"m.room.canonical_alias\",\"m.space.child\"],\"lazy_load_members\":true},\"timeline\":{\"limit\":20,\"lazy_load_members\":true},\"ephemeral\":{\"limit\":0,\"types\":[\"\"],\"lazy_load_members\":true},\"account_data\":{\"limit\":0,\"types\":[\"\"],\"lazy_load_members\":true}}}";
     static const char* CONTINUE_FILTER = "{\"presence\":{\"limit\":0,\"types\":[\"\"]},\"account_data\":{\"types\":[\"qm.emoji\",\"m.direct\"]},\"room\":{\"state\":{\"not_types\":[\"m.room.related_groups\",\"m.room.power_levels\",\"m.room.join_rules\",\"m.room.history_visibility\",\"m.room.canonical_alias\",\"m.space.child\"],\"lazy_load_members\":true},\"timeline\":{\"lazy_load_members\":true},\"ephemeral\":{\"limit\":0,\"types\":[\"\"],\"lazy_load_members\":true},\"account_data\":{\"types\":[\"m.fully_read\",\"m.tag\",\"qm.last_read_message_timestamp\"],\"lazy_load_members\":true}}}";
 
     static bool is_gpg_installed = false;
@@ -581,18 +581,6 @@ namespace QuickMedia {
         //fetched_messages_by_event_id.clear();
         //userdata = nullptr;
         //user_info_by_user_id.clear();
-        size_t i = 0;
-        for(auto it = messages.begin(); it != messages.end();) {
-            if((*it)->cache) {
-                message_by_event_id.erase((*it)->event_id);
-                it = messages.erase(it);
-                if(i <= messages_read_index)
-                    --messages_read_index;
-            } else {
-                ++it;
-            }
-            ++i;
-        }
         //messages.clear();
         //messages_read_index = 0;
         //message_by_event_id.clear();
@@ -629,11 +617,11 @@ namespace QuickMedia {
         rooms_page->add_body_item(body_item);
     }
 
-    void MatrixQuickMedia::leave_room(RoomData *room, LeaveType leave_type, const std::string &reason, bool is_cache) {
+    void MatrixQuickMedia::leave_room(RoomData *room, LeaveType leave_type, const std::string &reason) {
         room_body_item_by_room.erase(room);
         rooms_page->remove_body_item_by_room_id(room->id);
         room_tags_page->remove_body_item_by_room_id(room->id);
-        if(!is_cache && leave_type != LeaveType::LEAVE)
+        if(leave_type != LeaveType::LEAVE)
             show_notification("QuickMedia", reason);
     }
 
@@ -651,11 +639,11 @@ namespace QuickMedia {
         room_tags_page->remove_room_body_item_from_tag(it->second, tag);
     }
 
-    void MatrixQuickMedia::room_add_new_messages(RoomData *room, const Messages &messages, bool is_initial_sync, bool sync_is_cache, MessageDirection message_dir) {
+    void MatrixQuickMedia::room_add_new_messages(RoomData *room, const Messages &messages, bool is_initial_sync, MessageDirection message_dir) {
         bool is_window_focused = program->is_window_focused();
         RoomData *current_room = program->get_current_chat_room();
 
-        if(!sync_is_cache && message_dir == MessageDirection::AFTER) {
+        if(message_dir == MessageDirection::AFTER) {
             for(auto &message : messages) {
                 if(message->notification_mentions_me) {
                     std::string body = remove_reply_formatting(matrix, message->body);
@@ -679,7 +667,7 @@ namespace QuickMedia {
             }
         }
 
-        update_room_description(room, messages, is_initial_sync, sync_is_cache);
+        update_room_description(room, messages, is_initial_sync);
     }
 
     void MatrixQuickMedia::add_invite(const std::string &room_id, const Invite &invite) {
@@ -802,15 +790,6 @@ namespace QuickMedia {
             body->set_selected_item(found_item_index, false);
     }
 
-    void MatrixQuickMedia::clear_data() {
-        //room_body_item_by_room.clear();
-        //pending_room_messages.clear();
-        //rooms_page->clear_data();
-        //room_tags_page->clear_data();
-        invites_page->clear_data();
-        //unread_notifications.clear();
-    }
-
     static std::shared_ptr<Message> get_last_message_by_timestamp(const Messages &messages) {
         if(messages.empty())
             return nullptr;
@@ -870,7 +849,7 @@ namespace QuickMedia {
         return !get_config().matrix.gpg_user_id.empty() && text.find("-----BEGIN PGP MESSAGE-----") != std::string::npos && text.find("-----END PGP MESSAGE-----") != std::string::npos;
     }
 
-    void MatrixQuickMedia::update_room_description(RoomData *room, const Messages &new_messages, bool is_initial_sync, bool sync_is_cache) {
+    void MatrixQuickMedia::update_room_description(RoomData *room, const Messages &new_messages, bool is_initial_sync) {
         time_t read_marker_message_timestamp = 0;
         std::shared_ptr<UserInfo> me = matrix->get_me(room);
         std::string my_user_read_marker;
@@ -915,7 +894,7 @@ namespace QuickMedia {
         if(!room->body_item)
             return;
 
-        if(!sync_is_cache && last_unread_message) {
+        if(last_unread_message) {
             bool is_window_focused = program->is_window_focused();
             RoomData *current_room = program->get_current_chat_room();
             Body *chat_body = chat_page ? chat_page->chat_body : nullptr;
@@ -1038,12 +1017,6 @@ namespace QuickMedia {
         body->select_first_item();
     }
 
-    void MatrixRoomsPage::clear_data() {
-        body->clear_items();
-        if(current_chat_page)
-            current_chat_page->should_clear_data = true;
-    }
-
     PluginResult MatrixRoomTagsPage::submit(const SubmitArgs &args, std::vector<Tab> &result_tabs) {
         auto body = create_body(true);
         Body *body_ptr = body.get();
@@ -1131,13 +1104,6 @@ namespace QuickMedia {
         current_rooms_page = rooms_page;
     }
 
-    void MatrixRoomTagsPage::clear_data() {
-        tag_body_items_by_name.clear();
-        body->clear_items();
-        if(current_rooms_page)
-            current_rooms_page->clear_data();
-    }
-
     MatrixInvitesPage::MatrixInvitesPage(Program *program, Matrix *matrix, Body *body) : Page(program), matrix(matrix), body(body) {
 
     }
@@ -1198,12 +1164,6 @@ namespace QuickMedia {
         }
     }
 
-    void MatrixInvitesPage::clear_data() {
-        body->clear_items();
-        prev_invite_count = 0;
-        title = "Invites (0)";
-    }
-
     PluginResult MatrixSettingsPage::submit(const SubmitArgs &args, std::vector<Tab> &result_tabs) {
         if(args.url == "join") {
             result_tabs.push_back(Tab{create_body(), std::make_unique<MatrixRoomInputPage>(program, matrix), create_search_bar("Enter room id...", SEARCH_DELAY_FILTER)});
@@ -1677,43 +1637,7 @@ namespace QuickMedia {
         "M_MISSING_PARAM"
     };
 
-    static void remove_empty_fields_in_sync_rooms_response(rapidjson::Value &rooms_json) {
-        for(const char *member_name : {"join", "invite", "leave"}) {
-            auto join_it = rooms_json.FindMember(member_name);
-            if(join_it != rooms_json.MemberEnd() && join_it->value.IsObject() && join_it->value.MemberCount() == 0)
-                rooms_json.RemoveMember(join_it);
-        }
-    }
-
-    static void remove_empty_fields_in_sync_account_data_response(rapidjson::Value &account_data_json) {
-        for(const char *member_name : {"events"}) {
-            auto join_it = account_data_json.FindMember(member_name);
-            if(join_it != account_data_json.MemberEnd() && join_it->value.IsObject() && join_it->value.MemberCount() == 0)
-                account_data_json.RemoveMember(join_it);
-        }
-    }
-
-    static void remove_unused_sync_data_fields(rapidjson::Value &json_root) {
-        for(auto it = json_root.MemberBegin(); it != json_root.MemberEnd();) {
-            if(strcmp(it->name.GetString(), "account_data") == 0 && it->value.IsObject()) {
-                remove_empty_fields_in_sync_account_data_response(it->value);
-                if(it->value.MemberCount() == 0)
-                    it = json_root.RemoveMember(it);
-                else
-                    ++it;
-            } else if(strcmp(it->name.GetString(), "rooms") == 0 && it->value.IsObject()) {
-                // TODO: Call this, but dont remove our read marker (needed for notifications on mentions for example). Or maybe we can get it from "account_data"?
-                //remove_ephemeral_field_in_sync_rooms_response(rooms_it->value);
-                remove_empty_fields_in_sync_rooms_response(it->value);
-                if(it->value.MemberCount() == 0)
-                    it = json_root.RemoveMember(it);
-                else
-                    ++it;
-            } else {
-                it = json_root.EraseMember(it);
-            }
-        }
-    }
+    Matrix::Matrix(bool matrix_instance_already_running) : matrix_instance_already_running(matrix_instance_already_running) {}
 
     bool Matrix::start_sync(MatrixDelegate *delegate, bool &cached) {
         cached = true;
@@ -1734,31 +1658,39 @@ namespace QuickMedia {
 
         cached = (get_file_type(matrix_cache_dir) == FileType::REGULAR);
 
-        sync_is_cache = false;
         sync_running = true;
 
         load_silenced_invites();
         load_custom_emoji_from_cache();
 
         sync_thread = std::thread([this, matrix_cache_dir]() {
-            sync_is_cache = true;
-            FILE *sync_cache_file = fopen(matrix_cache_dir.data.c_str(), "rb");
-            if(sync_cache_file) {
+            FILE *sync_cache_file;
+            const rapidjson::Value *next_batch_json = nullptr;
+
+            load_qm_read_markers_from_account_data(); // TODO: Remove when https://github.com/matrix-org/synapse/issues/14444 is fixed, if ever.
+
+            std::ifstream sync_cache_file_stream;
+            sync_cache_file_stream.open(matrix_cache_dir.data.c_str(), std::ifstream::in | std::ifstream::binary);
+            if(sync_cache_file_stream.good() && get_file_type(get_cache_dir().join("matrix").join("updated-cache-version1")) == FileType::REGULAR) {
                 rapidjson::Document doc;
-                char read_buffer[8192];
-                rapidjson::FileReadStream is(sync_cache_file, read_buffer, sizeof(read_buffer));
-                while(true) {
-                    rapidjson::ParseResult parse_result = doc.ParseStream<rapidjson::kParseStopWhenDoneFlag>(is);
+                std::string line;
+                while(std::getline(sync_cache_file_stream, line)) {
+                    rapidjson::ParseResult parse_result = doc.Parse<rapidjson::kParseStopWhenDoneFlag>(line.c_str(), line.size());
                     if(parse_result.IsError())
-                        break;
-                    if(parse_sync_response(doc, false, false) != PluginResult::OK)
+                        continue; // This should NEVER happen. Do initial sync if it does and remove cache? :( TODO
+
+                    if(parse_sync_response(doc, false) != PluginResult::OK)
                         fprintf(stderr, "Failed to parse cached sync response\n");
+
+                    next_batch_json = &GetMember(doc, "next_batch");
+                    if(next_batch_json->IsString()) {
+                        set_next_batch(next_batch_json->GetString());
+                        //fprintf(stderr, "Matrix: next batch: %s\n", next_batch.c_str());
+                    }
                 }
-                fclose(sync_cache_file);
+                malloc_trim(0);
             }
-            sync_is_cache = false;
-
-            load_qm_read_markers_from_account_data(); // TODO: Remove when https://github.com/matrix-org/synapse/issues/14444 is fixed, if ever.
+            sync_cache_file_stream.close();
 
             // Filter with account data
             // {"presence":{"limit":0,"types":[""]},"account_data":{"not_types":["im.vector.setting.breadcrumbs","m.push_rules","im.vector.setting.allowed_widgets","io.element.recent_emoji"]},"room":{"state":{"limit":1,"not_types":["m.room.related_groups","m.room.power_levels","m.room.join_rules","m.room.history_visibility"],"lazy_load_members":true},"timeline":{"limit":3,"lazy_load_members":true},"ephemeral":{"limit":0,"types":[""],"lazy_load_members":true},"account_data":{"limit":1,"types":["m.fully_read"],"lazy_load_members":true}}}
@@ -1777,16 +1709,35 @@ namespace QuickMedia {
             else
                 filter = FILTER;
 #endif
-            std::string filter_encoded = url_param_encode(INITIAL_FILTER);
-
             std::vector<CommandArg> additional_args = {
                 { "-H", "Authorization: Bearer " + access_token },
                 { "-m", "35" }
             };
 
-            const rapidjson::Value *next_batch_json;
+            next_batch_json = nullptr;
             PluginResult result;
-            bool initial_sync = true;
+            bool initial_sync = next_batch.empty();
+            bool first_sync = true;
+
+            notification_thread = std::thread([this]() {
+                get_previous_notifications([this](const MatrixNotification &notification) {
+                    if(notification.read)
+                        return;
+
+                    MatrixDelegate *delegate = this->delegate;
+                    ui_thread_tasks.push([delegate, notification] {
+                        delegate->add_unread_notification(std::move(notification));
+                    });
+                });
+                finished_fetching_notifications = true;
+            });
+
+            std::string filter_encoded;
+            if(initial_sync)
+                filter_encoded = url_param_encode(INITIAL_FILTER);
+            else
+                filter_encoded = url_param_encode(CONTINUE_FILTER);
+
             while(sync_running) {
                 char url[2048];
                 if(next_batch.empty())
@@ -1821,10 +1772,7 @@ namespace QuickMedia {
                     }
                 }
 
-                if(next_batch.empty())
-                    clear_sync_cache_for_new_sync();
-
-                result = parse_sync_response(json_root, false, initial_sync);
+                result = parse_sync_response(json_root, initial_sync);
                 if(result != PluginResult::OK) {
                     fprintf(stderr, "Failed to parse sync response\n");
                     initial_sync = false;
@@ -1842,50 +1790,9 @@ namespace QuickMedia {
                     goto sync_end;
                 }
 
-                if(initial_sync) {
-                    notification_thread = std::thread([this]() {
-                        get_previous_notifications([this](const MatrixNotification &notification) {
-                            if(notification.read)
-                                return;
-
-                            MatrixDelegate *delegate = this->delegate;
-                            ui_thread_tasks.push([delegate, notification] {
-                                delegate->add_unread_notification(std::move(notification));
-                            });
-                        });
-
-                        finished_fetching_notifications = true;
-
-                        {
-                            std::vector<CommandArg> additional_args = {
-                                { "-H", "Authorization: Bearer " + access_token },
-                                { "-m", "35" }
-                            };
-
-                            char url[1024];
-                            std::string filter_encoded = url_param_encode(ADDITIONAL_MESSAGES_FILTER);
-                            snprintf(url, sizeof(url), "%s/_matrix/client/r0/sync?filter=%s&timeout=0", homeserver.c_str(), filter_encoded.c_str());
-
-                            rapidjson::Document json_root;
-                            std::string err_msg;
-                            DownloadResult download_result = download_json(json_root, url, additional_args, true, &err_msg);
-                            if(download_result != DownloadResult::OK) {
-                                fprintf(stderr, "/sync for additional messages failed\n");
-                                return;
-                            }
-
-                            // TODO: Test?
-                            //if(next_batch.empty())
-                            //    clear_sync_cache_for_new_sync();
-
-                            additional_messages_queue.pop_wait();
-                            parse_sync_response(json_root, true, false);
-                        }
-                    });
-
+                if(first_sync) {
+                    first_sync = false;
                     filter_encoded = url_param_encode(CONTINUE_FILTER);
-                    additional_messages_queue.push(true);
-                    malloc_trim(0);
                 }
 
 #if 0
@@ -1895,18 +1802,33 @@ namespace QuickMedia {
                 }
 #endif
 
-                // TODO: Circulate file
-                sync_cache_file = fopen(matrix_cache_dir.data.c_str(), initial_sync ? "wb" : "ab");
-                initial_sync = false;
-                if(sync_cache_file) {
-                    if(json_root.IsObject()) {
-                        char buffer[4096];
-                        rapidjson::FileWriteStream file_write_stream(sync_cache_file, buffer, sizeof(buffer));
-                        rapidjson::Writer<rapidjson::FileWriteStream> writer(file_write_stream);
-                        remove_unused_sync_data_fields(json_root);
-                        json_root.Accept(writer);
+                // TODO: Use a NoSQL database.
+                // TODO: Remove very old cache.
+                // TODO: Find a way to remove this? this makes sync work like other clients but we dont want that!
+                //       If the last sync was long ago then it has to sync ALL messages again. Maybe check if sync file
+                //       is XX days old and then ignore it?
+                // TODO: Remove this matrix_instance_already_running check when the matrix sync is moved to a daemon.
+                //       Then the daemon will do the sync and matrix processes will ask that daemon for the cached data
+                //       and fetch previous messages etc themselves.
+                if(!matrix_instance_already_running) {
+                    sync_cache_file = fopen(matrix_cache_dir.data.c_str(), initial_sync ? "wb" : "ab");
+                    initial_sync = false;
+                    if(sync_cache_file) {
+                        if(json_root.IsObject()) {
+                            rapidjson::StringBuffer buffer;
+                            rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
+                            json_root.Accept(writer);
+
+                            std::string json_data(buffer.GetString(), buffer.GetSize());
+                            string_replace_all(json_data, '\n', ' ');
+                            json_data += '\n';
+
+                            fwrite(json_data.data(), 1, json_data.size(), sync_cache_file);
+                            file_overwrite(get_cache_dir().join("matrix").join("updated-cache-version1"), "1"); // To make sure the cache format is up to date
+                            malloc_trim(0);
+                        }
+                        fclose(sync_cache_file);
                     }
-                    fclose(sync_cache_file);
                 }
 
                 sync_end:
@@ -1925,13 +1847,6 @@ namespace QuickMedia {
             program_kill_in_thread(sync_thread.get_id());
             sync_thread.join();
         }
-
-        if(sync_additional_messages_thread.joinable()) {
-            program_kill_in_thread(sync_additional_messages_thread.get_id());
-            additional_messages_queue.close();
-            sync_additional_messages_thread.join();
-            additional_messages_queue.restart();
-        }
         
         if(notification_thread.joinable()) {
             program_kill_in_thread(notification_thread.get_id());
@@ -2113,12 +2028,12 @@ namespace QuickMedia {
         }
     }
 
-    PluginResult Matrix::parse_sync_response(const rapidjson::Document &root, bool is_additional_messages_sync, bool initial_sync) {
+    PluginResult Matrix::parse_sync_response(const rapidjson::Document &root, bool initial_sync) {
         if(!root.IsObject())
             return PluginResult::ERR;
 
         const rapidjson::Value &rooms_json = GetMember(root, "rooms");
-        parse_sync_room_data(rooms_json, is_additional_messages_sync, initial_sync);
+        parse_sync_room_data(rooms_json, initial_sync);
 
         const rapidjson::Value &account_data_json = GetMember(root, "account_data");
         parse_sync_account_data(account_data_json);
@@ -2300,7 +2215,7 @@ namespace QuickMedia {
         return PluginResult::OK;
     }
 
-    PluginResult Matrix::parse_sync_room_data(const rapidjson::Value &rooms_json, bool is_additional_messages_sync, bool initial_sync) {
+    PluginResult Matrix::parse_sync_room_data(const rapidjson::Value &rooms_json, bool initial_sync) {
         if(!rooms_json.IsObject())
             return PluginResult::OK;
 
@@ -2342,16 +2257,9 @@ namespace QuickMedia {
 
                 const rapidjson::Value &timeline_json = GetMember(it.value, "timeline");
                 if(timeline_json.IsObject()) {
-                    if(is_additional_messages_sync) {
-                        // This may be non-existent if this is the first event in the room
-                        const rapidjson::Value &prev_batch_json = GetMember(timeline_json, "prev_batch");
-                        if(prev_batch_json.IsString())
-                            room->set_prev_batch(prev_batch_json.GetString());
-                    }
-
                     bool has_unread_notifications = false;
                     const rapidjson::Value &unread_notification_json = GetMember(it.value, "unread_notifications");
-                    if(unread_notification_json.IsObject() && !is_additional_messages_sync && !sync_is_cache) {
+                    if(unread_notification_json.IsObject()) {
                         const rapidjson::Value &highlight_count_json = GetMember(unread_notification_json, "highlight_count");
                         if(highlight_count_json.IsInt64() && (highlight_count_json.GetInt64() > 0 || initial_sync)) {
                             room->unread_notification_count = highlight_count_json.GetInt64();
@@ -2368,7 +2276,7 @@ namespace QuickMedia {
                     if(account_data_json.IsObject()) {
                         const rapidjson::Value &events_json = GetMember(account_data_json, "events");
                         auto me = get_me(room);
-                        events_set_user_read_marker(events_json, room, me, is_additional_messages_sync);
+                        events_set_user_read_marker(events_json, room, me);
                     }
 
                     if(is_new_room)
@@ -2381,7 +2289,7 @@ namespace QuickMedia {
                     if(account_data_json.IsObject()) {
                         const rapidjson::Value &events_json = GetMember(account_data_json, "events");
                         auto me = get_me(room);
-                        events_set_user_read_marker(events_json, room, me, is_additional_messages_sync);
+                        events_set_user_read_marker(events_json, room, me);
                     }
 
                     if(is_new_room)
@@ -2410,20 +2318,18 @@ namespace QuickMedia {
             }
         }
 
-        if(!is_additional_messages_sync) {
-            const rapidjson::Value &leave_json = GetMember(rooms_json, "leave");
-            remove_rooms(leave_json);
+        const rapidjson::Value &leave_json = GetMember(rooms_json, "leave");
+        remove_rooms(leave_json);
 
-            const rapidjson::Value &invite_json = GetMember(rooms_json, "invite");
-            add_invites(invite_json);
-        }
+        const rapidjson::Value &invite_json = GetMember(rooms_json, "invite");
+        add_invites(invite_json);
 
         if(initial_sync) {
             std::lock_guard<std::recursive_mutex> lock(room_data_mutex);
             for(auto &room : rooms) {
                 if(existing_rooms.find(room.get()) == existing_rooms.end()) {
                     RoomData *room_p = room.get();
-                    ui_thread_tasks.push([this, room_p]{ delegate->leave_room(room_p, LeaveType::LEAVE, "", true); });
+                    ui_thread_tasks.push([this, room_p]{ delegate->leave_room(room_p, LeaveType::LEAVE, ""); });
                     remove_room(room->id);
                 }
             }
@@ -2549,7 +2455,7 @@ namespace QuickMedia {
         return user_info;
     }
 
-    void Matrix::events_set_user_read_marker(const rapidjson::Value &events_json, RoomData *room_data, std::shared_ptr<UserInfo> &me, bool is_additional_messages_sync) {
+    void Matrix::events_set_user_read_marker(const rapidjson::Value &events_json, RoomData *room_data, std::shared_ptr<UserInfo> &me) {
         assert(me); // TODO: Remove read marker from user and set it for the room instead. We need that in the matrix pages also
         if(!events_json.IsArray() || !me)
             return;
@@ -2571,8 +2477,7 @@ namespace QuickMedia {
                 if(!event_id_json.IsString())
                     continue;
                 
-                if(!sync_is_cache && !is_additional_messages_sync)
-                    room_data->set_user_read_marker(me, std::string(event_id_json.GetString(), event_id_json.GetStringLength()));
+                room_data->set_user_read_marker(me, std::string(event_id_json.GetString(), event_id_json.GetStringLength()));
             } else if(strcmp(type_json.GetString(), "qm.last_read_message_timestamp") == 0) { // TODO: Remove qm.last_read_message_timestamp in room level eventually when everybody has data in global level
                 const rapidjson::Value &content_json = GetMember(event_json, "content");
                 if(!content_json.IsObject())
@@ -2711,7 +2616,7 @@ namespace QuickMedia {
         new_messages.push_back(std::move(message));
         room_data->append_messages(new_messages);
         ui_thread_tasks.push([this, room_data, new_messages{std::move(new_messages)}]{
-            delegate->room_add_new_messages(room_data, new_messages, false, false, MessageDirection::AFTER);
+            delegate->room_add_new_messages(room_data, new_messages, false, MessageDirection::AFTER);
         });
     }
 
@@ -2726,10 +2631,8 @@ namespace QuickMedia {
 
         for(const rapidjson::Value &event_item_json : events_json.GetArray()) {
             std::shared_ptr<Message> new_message = parse_message_event(event_item_json, room_data);
-            if(new_message) {
-                new_message->cache = sync_is_cache;
+            if(new_message)
                 new_messages.push_back(std::move(new_message));
-            }
         }
 
         if(new_messages.empty())
@@ -2766,10 +2669,9 @@ namespace QuickMedia {
             }
         }
 
-        bool cache_sync = sync_is_cache;
         bool is_initial_sync = next_batch.empty();
-        ui_thread_tasks.push([this, room_data, cache_sync, new_messages{std::move(new_messages)}, is_initial_sync, message_dir]{
-            delegate->room_add_new_messages(room_data, new_messages, is_initial_sync, cache_sync, message_dir);
+        ui_thread_tasks.push([this, room_data, new_messages{std::move(new_messages)}, is_initial_sync, message_dir]{
+            delegate->room_add_new_messages(room_data, new_messages, is_initial_sync, message_dir);
         });
 
         return num_new_messages;
@@ -3775,7 +3677,6 @@ namespace QuickMedia {
                     invite.room_avatar_url = room->get_avatar_url();
                     invite.invited_by = invited_by;
                     invite.timestamp = timestamp;
-                    invite.new_invite = !sync_is_cache;
 
                     std::string room_id_str(room_id.GetString(), room_id.GetStringLength());
                     if(set_invite(room_id_str, invite))
@@ -3861,8 +3762,7 @@ namespace QuickMedia {
                 if(!reason_str.empty())
                     desc += ", reason: " + reason_str;
 
-                const bool is_cache = sync_is_cache;
-                ui_thread_tasks.push([this, room, leave_type, desc{std::move(desc)}, is_cache]{ delegate->leave_room(room, leave_type, desc, is_cache); });
+                ui_thread_tasks.push([this, room, leave_type, desc{std::move(desc)}]{ delegate->leave_room(room, leave_type, desc); });
                 remove_room(room_id_str);
                 break;
             }
@@ -5523,7 +5423,7 @@ namespace QuickMedia {
         if(download_result == DownloadResult::OK) {
             RoomData *room = get_room_by_id(room_id);
             if(room) {
-                ui_thread_tasks.push([this, room]{ delegate->leave_room(room, LeaveType::LEAVE, "", false); });
+                ui_thread_tasks.push([this, room]{ delegate->leave_room(room, LeaveType::LEAVE, ""); });
                 remove_room(room_id);
             }
         }
@@ -5887,18 +5787,6 @@ namespace QuickMedia {
         return next_notifications_token;
     }
 
-    void Matrix::clear_sync_cache_for_new_sync() {
-        std::lock_guard<std::recursive_mutex> room_data_lock(room_data_mutex);
-        std::lock_guard<std::mutex> invites_lock(invite_mutex);
-        for(auto &room : rooms) {
-            room->clear_data();
-        }
-        // We intentionally dont clear |rooms| here because we want the objects inside it to still be valid. TODO: Clear |rooms| here
-        //room_data_by_id.clear();
-        invites.clear();
-        ui_thread_tasks.push([this]{ delegate->clear_data(); });
-    }
-
     std::shared_ptr<UserInfo> Matrix::get_user_by_id(RoomData *room, const std::string &user_id, bool *is_new_user, bool create_if_not_found) {
         auto user = room->get_user_by_id(user_id);
         if(user) {
diff --git a/src/plugins/utils/UniqueProcess.cpp b/src/plugins/utils/UniqueProcess.cpp
new file mode 100644
index 0000000..d2025f5
--- /dev/null
+++ b/src/plugins/utils/UniqueProcess.cpp
@@ -0,0 +1,111 @@
+#include "../../../plugins/utils/UniqueProcess.hpp"
+#include "../../../include/Storage.hpp"
+#include <stdio.h>
+#include <limits.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <errno.h>
+
+namespace QuickMedia {
+    bool is_quickmedia_instance_already_running(const char *sock_file_dir, const char *plugin_name) {
+        char sock_file[PATH_MAX];
+        snprintf(sock_file, sizeof(sock_file), "%s/quickmedia.%s.sock", sock_file_dir, plugin_name);
+
+        std::string resolved_path;
+        if(file_get_content(sock_file, resolved_path) != 0)
+            return false;
+        
+        resolved_path.resize(108); // sizeof(addr.sun_path) is 108
+
+        int fd = socket(AF_UNIX, SOCK_STREAM, 0);
+        if(fd == -1) {
+            fprintf(stderr, "Error: failed to create unix domain socket, error: %s\n", strerror(errno));
+            return true;
+        }
+
+        fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
+
+        struct sockaddr_un addr;
+        memset(&addr, 0, sizeof(addr));
+        addr.sun_family = AF_UNIX;
+        strcpy(addr.sun_path, resolved_path.c_str());
+
+        bool running = connect(fd, (struct sockaddr*)&addr, sizeof(addr)) == 0;
+        int err = errno;
+        if(err == EAGAIN)
+            running = true;
+        else if(err == ENOENT)
+            running = false;
+        close(fd);
+        return running;
+    }
+
+    bool set_quickmedia_instance_unique(const char *sock_file_dir, const char *plugin_name) {
+        char socket_file[] = "/tmp/quickmedia.XXXXXX";
+        int tmp_file_fd = mkstemp(socket_file);
+        if(tmp_file_fd == -1) {
+            fprintf(stderr, "Error: failed to create temporary file for unix domain socket, error: %s\n", strerror(errno));
+            return false;
+        }
+        unlink(socket_file);
+        close(tmp_file_fd);
+
+        int fd = socket(AF_UNIX, SOCK_STREAM, 0);
+        if(fd == -1) {
+            fprintf(stderr, "Error: failed to create unix domain socket, error: %s\n", strerror(errno));
+            return false;
+        }
+
+        struct sockaddr_un addr;
+        memset(&addr, 0, sizeof(addr));
+        addr.sun_family = AF_UNIX;
+        strcpy(addr.sun_path, socket_file);
+
+        if(bind(fd, (struct sockaddr*)&addr, sizeof(addr)) == -1) {
+            fprintf(stderr, "Error: failed to bind unix domain socket, error: %s\n", strerror(errno));
+            unlink(socket_file);
+            close(fd);
+            return false;
+        }
+
+        if(listen(fd, 0) == -1) {
+            fprintf(stderr, "Error: failed to listen to unix domain socket, error: %s\n", strerror(errno));
+            unlink(socket_file);
+            close(fd);
+            return false;
+        }
+
+        char sock_file[PATH_MAX];
+        snprintf(sock_file, sizeof(sock_file), "%s/quickmedia.%s.sock", sock_file_dir, plugin_name);
+        bool success = file_overwrite(sock_file, socket_file) == 0;
+        if(!success) {
+            fprintf(stderr, "Error: failed to create %s unix domain socket link file\n", sock_file);
+            unlink(socket_file);
+            close(fd);
+        }
+        return success;
+    }
+
+    void remove_quickmedia_instance_lock(const char *sock_file_dir, const char *plugin_name) {
+        char sock_file[PATH_MAX];
+        snprintf(sock_file, sizeof(sock_file), "%s/quickmedia.%s.sock", sock_file_dir, plugin_name);
+
+        std::string resolved_path;
+        if(file_get_content(sock_file, resolved_path) != 0) {
+            unlink(sock_file);
+            return;
+        }
+        
+        resolved_path.resize(108); // sizeof(addr.sun_path) is 108
+
+        if(resolved_path.size() < 4 || memcmp(resolved_path.data(), "/tmp", 4) != 0) {
+            unlink(sock_file);
+            return;
+        }
+        
+        unlink(sock_file);
+        unlink(resolved_path.c_str());
+    }
+}
\ No newline at end of file