Messenger: hopefully handle joins correctly

2 files changed, 114 insertions, 67 deletions

diff --git a/connector/external/external.go b/connector/external/external.go
index 9a4137d..6d25230 100644
--- a/connector/external/external.go
+++ b/connector/external/external.go
@@ -111,7 +111,7 @@ func (ext *External) Configure(c Configuration) error {
 
 	ext.generation += 1
 
-	ext.handlerChan = make(chan *extMessageWithData)
+	ext.handlerChan = make(chan *extMessageWithData, 1000)
 	go ext.handlerLoop(ext.generation)
 
 	err = ext.setupProc()
diff --git a/external/messenger.py b/external/messenger.py
index cb48410..9728c9d 100755
--- a/external/messenger.py
+++ b/external/messenger.py
@@ -95,52 +95,12 @@ class InitialSyncThread(threading.Thread):
         self.threads = threads
 
     def run(self):
-        sys.stderr.write("fb thread list: {}\n".format(self.threads))
+        sys.stderr.write("(python) fb thread list: {}\n".format(self.threads))
 
         for thread in self.threads:
-            sys.stderr.write("fb thread: {}\n".format(thread))
-            if thread.type == ThreadType.GROUP:
-                members = self.client.fetchAllUsersFromThreads([thread])
+            sys.stderr.write("(python) fb thread: {}\n".format(thread))
+            self.bridge.setup_joined_thread(thread)
 
-                self.bridge.write({
-                    "_type": JOINED,
-                    "room": thread.uid,
-                })
-
-                self.bridge.send_room_info(thread, members)
-                self.bridge.send_room_members(thread, members)
-
-            self.backlog_room(thread)
-
-
-    def backlog_room(self, thread):
-        prev_last_seen = self.bridge.cache_get("last_seen_%s"%thread.uid)
-        if prev_last_seen == "":
-            prev_last_seen = None
-
-        messages = []
-        found = False
-        while not found:
-            before = None
-            if len(messages) > 0:
-                before = messages[-1].timestamp
-            page = self.client.fetchThreadMessages(thread.uid, before=before, limit=20)
-            for m in page:
-                if m.uid == prev_last_seen or len(messages) > self.bridge.init_backlog_length:
-                    found = True
-                    break
-                else:
-                    messages.append(m)
-
-        for m in reversed(messages):
-            if m.text is None:
-                m.text = ""
-            m.text = "[{}] {}".format(
-                    time.strftime("%Y-%m-%d %H:%M %Z", time.localtime(float(m.timestamp)/1000)).strip(),
-                    m.text)
-            self.bridge.onMessage(thread_id=thread.uid,
-                                  thread_type=thread.type,
-                                  message_object=m)
 
 class ClientListenThread(threading.Thread):
     def __init__(self, client, *args, **kwargs):
@@ -149,7 +109,7 @@ class ClientListenThread(threading.Thread):
         self.client = client
 
     def run(self):
-        sys.stderr.write("Start client.listen()\n")
+        sys.stderr.write("(python messenger) Start client.listen()\n")
         self.client.listen()
 
 
@@ -159,7 +119,8 @@ class MessengerBridge:
     def __init__(self):
         self.rev_uid = {}
         self.uid_map = {}
-        self.joined_map = {}
+        self.others_joined_map = {}
+        self.my_joined_rooms = {}
         self.init_backlog_length = 100
 
     def getUserId(self, user):
@@ -215,7 +176,7 @@ class MessengerBridge:
             try:
                 line = sys.stdin.readline()
             except KeyboardInterrupt:
-                sys.stderr.write("(messenger) shutting down")
+                sys.stderr.write("(python messenger) shutting down")
                 self.close()
                 time.sleep(5)
                 sys.exit(0)
@@ -230,7 +191,7 @@ class MessengerBridge:
                 if "_type" not in rep:
                     rep["_type"] = REP_OK
             except Exception as e:
-                sys.stderr.write("{}\n".format(traceback.format_exc()))
+                sys.stderr.write("(python) {}\n".format(traceback.format_exc()))
                 rep = {
                         "_type": REP_ERROR,
                         "error": "{}".format(e)
@@ -292,8 +253,19 @@ class MessengerBridge:
             userId = self.getUserIdFromUid(self.client.uid)
             return {"_type": REP_OK, "user": userId}
 
-        elif ty == INVITE and cmd["room"] == "":
-            return {"_type": REP_OK}
+        elif ty == JOIN:
+            self.ensure_i_joined(cmd["room"])
+
+        elif ty == LEAVE:
+            thread_id = cmd["room"]
+            self.client.removeUserFromGroup(self.client.uid, thread_id)
+            if thread_id in self.my_joined_rooms:
+                del self.my_joined_rooms[thread_id]
+
+        elif ty == INVITE:
+            if cmd["room"] != "":
+                uid = self.revUserId(cmd["user"])
+                self.client.addUsersToGroup([uid], cmd["room"])
 
         elif ty == SEND:
             event = cmd["data"]
@@ -318,7 +290,7 @@ class MessengerBridge:
                         msg_id = self.client.send(msg, thread_id=event["room"], thread_type=ThreadType.GROUP)
                 elif event["recipient"] != "":
                     uid = self.revUserId(event["recipient"])
-                    sys.stderr.write("Sending to {}\n".format(uid))
+                    sys.stderr.write("(python) Sending to {}\n".format(uid))
                     if len(attachments) > 0:
                         msg_id = self.client.sendRemoteFiles(attachments, message=msg, thread_id=uid, thread_type=ThreadType.USER)
                     else:
@@ -344,13 +316,40 @@ class MessengerBridge:
         q = queue.Queue(1)
         self.cache_gets[num] = q
         self.write({"_type": CACHE_GET, "_id": num, "key": key})
-        rep = q.get(block=True, timeout=30)
+        try:
+            rep = q.get(block=True, timeout=30)
+        except queue.Empty:
+            rep = ""
         del self.cache_gets[num]
         return rep
 
     def cache_put(self, key, value):
         self.write({"_type": CACHE_PUT, "key": key, "value": value})
 
+    # ---- Info sync ----
+
+    def ensure_i_joined(self, thread_id):
+        if thread_id not in self.my_joined_rooms:
+            thread = self.client.fetchThreadInfo(thread_id)[thread_id]
+            self.setup_joined_thread(thread)
+
+    def setup_joined_thread(self, thread):
+        self.my_joined_rooms[thread.uid] = True
+
+        if thread.type == ThreadType.GROUP:
+            members = self.client.fetchAllUsersFromThreads([thread])
+
+            self.write({
+                "_type": JOINED,
+                "room": thread.uid,
+            })
+
+            self.send_room_info(thread, members)
+            self.send_room_members(thread, members)
+
+        self.backlog_room(thread)
+
+
     def send_room_info(self, thread, members):
         room_info = {}
         if thread.name is not None:
@@ -378,12 +377,41 @@ class MessengerBridge:
 
     def send_room_members(self, thread, members):
         for member in members:
-            sys.stderr.write("fb thread member: {}\n".format(member))
+            sys.stderr.write("(python) fb thread member: {}\n".format(member))
             self.ensureJoined(self.getUserId(member), thread.uid)
 
+    def backlog_room(self, thread):
+        prev_last_seen = self.cache_get("last_seen_%s"%thread.uid)
+        if prev_last_seen == "":
+            prev_last_seen = None
+
+        messages = []
+        found = False
+        while not found:
+            before = None
+            if len(messages) > 0:
+                before = messages[-1].timestamp
+            page = self.client.fetchThreadMessages(thread.uid, before=before, limit=20)
+            for m in page:
+                if m.uid == prev_last_seen or len(messages) > self.init_backlog_length:
+                    found = True
+                    break
+                else:
+                    messages.append(m)
+
+        for m in reversed(messages):
+            if m.text is None:
+                m.text = ""
+            m.text = "[{}] {}".format(
+                    time.strftime("%Y-%m-%d %H:%M %Z", time.localtime(float(m.timestamp)/1000)).strip(),
+                    m.text)
+            self.onMessage(thread_id=thread.uid,
+                                  thread_type=thread.type,
+                                  message_object=m)
+
     def ensureJoined(self, userId, room):
         key = "{}--{}".format(userId, room)
-        if not key in self.joined_map:
+        if not key in self.others_joined_map:
             self.write({
                 "_type": EVENT,
                 "data": {
@@ -392,9 +420,13 @@ class MessengerBridge:
                     "room": room,
                 }
             })
-            self.joined_map[key] = True
+            self.others_joined_map[key] = True
+
+    # ---- Event handlers ----
 
     def onMessage(self, thread_id, thread_type, message_object, **kwargs):
+        self.ensure_i_joined(thread_id)
+
         if message_object.author == self.client.uid:
             # Ignore our own messages
             return
@@ -442,29 +474,44 @@ class MessengerBridge:
             event["room"] = thread_id
             self.ensureJoined(author, thread_id)
 
-        self.write({"_type": EVENT, "data": event})
+        if event["text"] != "" or len(event["attachments"]) > 0:
+            self.write({"_type": EVENT, "data": event})
 
         self.cache_put("last_seen_%s"%thread_id, message_object.uid)
 
     def onPeopleAdded(self, added_ids, thread_id, *args, **kwargs):
         for user_id in added_ids:
-            self.ensureJoined(self.getUserIdFromUid(user_id), thread_id)
+            if user_id == self.client.uid:
+                self.ensure_i_joined(thread_id)
+            else:
+                self.ensureJoined(self.getUserIdFromUid(user_id), thread_id)
 
     def onPersonRemoved(self, removed_id, thread_id, *args, **kwargs):
-        userId = self.getUserIdFromUid(removed_id),
-        self.write({
-            "_type": EVENT,
-            "data": {
-                "type": EVENT_JOIN,
-                "author": userId,
+        if removed_id == self.client.uid:
+            self.write({
+                "_type": LEFT,
                 "room": thread_id,
-            }
-        })
-        del self.joined_map["{}--{}".format(userId, thread_id)]
+            })
+            if thread_id in self.my_joined_rooms:
+                del self.my_joined_rooms[thread_id]
+        else:
+            userId = self.getUserIdFromUid(removed_id),
+            self.write({
+                "_type": EVENT,
+                "data": {
+                    "type": EVENT_JOIN,
+                    "author": userId,
+                    "room": thread_id,
+                }
+            })
+            map_key = "{}--{}".format(userId, thread_id)
+            if map_key in self.others_joined_map:
+                del self.others_joined_map[map_key]
 
     def onTitleChange(self, author_id, new_title, thread_id, thread_type, *args, **kwargs):
+        self.ensure_i_joined(thread_id)
         if thread_type == ThreadType.GROUP:
-            self.bridge.write({
+            self.write({
                 "_type": ROOM_INFO_UPDATED,
                 "room": thread_id,
                 "data": {"name": new_title},