Add missed message backfilling

ROADMAP.md

@@ -31,7 +31,7 @@
   * [x] Message history
     * [x] Manually (`!tg backfill`)
     * [x] Automatically when creating portal
-    * [ ] Automatically for missed messages
+    * [x] Automatically for missed messages
   * [x] Avatars
   * [x] Presence
   * [x] Typing notifications

mautrix_telegram/commands/telegram/misc.py

@@ -185,8 +185,10 @@ async def sync(evt: CommandEvent) -> EventID:
         sync_only = None
 
     if not sync_only or sync_only == "chats":
-        await evt.sender.sync_dialogs(synchronous_create=True)
+        await evt.reply("Synchronizing chats...")
+        await evt.sender.sync_dialogs()
     if not sync_only or sync_only == "contacts":
+        await evt.reply("Synchronizing contacts...")
         await evt.sender.sync_contacts()
     if not sync_only or sync_only == "me":
         await evt.sender.update_info()

mautrix_telegram/portal/metadata.py

@@ -223,8 +223,8 @@ class PortalMetadata(BasePortal, ABC):
             await self.main_intent.get_joined_members(self.mxid)
 
     async def create_matrix_room(self, user: 'AbstractUser', entity: Union[TypeChat, User] = None,
-                                 invites: InviteList = None, update_if_exists: bool = True,
-                                 synchronous: bool = False) -> Optional[str]:
+                                 invites: InviteList = None, update_if_exists: bool = True
+                                 ) -> Optional[RoomID]:
         if self.mxid:
             if update_if_exists:
                 if not entity:
@@ -234,10 +234,7 @@ class PortalMetadata(BasePortal, ABC):
                         self.log.exception(f"Failed to get entity through {user.tgid} for update")
                         return self.mxid
                 update = self.update_matrix_room(user, entity, self.peer_type == "user")
-                if synchronous:
-                    await update
-                else:
-                    asyncio.ensure_future(update, loop=self.loop)
+                self.loop.create_task(update)
                 await self.invite_to_matrix(invites or [])
             return self.mxid
         async with self._room_create_lock:

mautrix_telegram/portal/telegram.py

@@ -421,7 +421,7 @@ class PortalTelegram(BasePortal, ABC):
         }
 
     async def backfill(self, source: 'AbstractUser', is_initial: bool = False,
-                       limit: Optional[int] = None) -> None:
+                       limit: Optional[int] = None, last_id: Optional[int] = None) -> None:
         limit = limit or (config["bridge.backfill.initial_limit"] if is_initial
                           else config["bridge.backfill.missed_limit"])
         if limit == 0:
@@ -429,18 +429,23 @@ class PortalTelegram(BasePortal, ABC):
         last = DBMessage.find_last(self.mxid, (source.tgid if self.peer_type != "channel"
                                                else self.tgid))
         min_id = last.tgid if last else 0
-        message = (await source.client.get_messages(self.peer, limit=1))[0]
+        if last_id is None:
+            message = (await source.client.get_messages(self.peer, limit=1))[0]
+            last_id = message.id
+        if last_id <= min_id:
+            # Nothing to backfill
+            return
         if limit < 0:
             limit = None
-            self.log.debug(f"Backfilling approximately {message.id - min_id} messages "
+            self.log.debug(f"Backfilling approximately {last_id - min_id} messages "
                            f"through {source.mxid}")
         elif self.peer_type == "channel":
             # This is a channel or supergroup, so we'll backfill messages based on the ID.
             # There are some cases, such as deleted messages, where this may backfill less
             # messages than the limit.
-            min_id = max(message.id - limit, min_id)
+            min_id = max(last_id - limit, min_id)
             limit = None
-            self.log.debug(f"Backfilling messages after ID {min_id} (last message: {message.id}) "
+            self.log.debug(f"Backfilling messages after ID {min_id} (last message: {last_id}) "
                            f"through {source.mxid}")
         else:
             # Private chats and normal groups don't have their own message ID namespace,

mautrix_telegram/user.py

@@ -21,6 +21,7 @@ import asyncio
 from telethon.tl.types import (TypeUpdate, UpdateNewMessage, UpdateNewChannelMessage, PeerUser,
                                UpdateShortChatMessage, UpdateShortMessage, User as TLUser, Chat,
                                ChatForbidden)
+from telethon.tl.custom import Dialog
 from telethon.tl.types.contacts import ContactsNotModified
 from telethon.tl.functions.contacts import GetContactsRequest, SearchRequest
 from telethon.tl.functions.account import UpdateStatusRequest
@@ -333,12 +334,13 @@ class User(AbstractUser, BaseUser):
 
         return await self._search_remote(query), True
 
-    async def sync_dialogs(self, synchronous_create: bool = False) -> None:
+    async def sync_dialogs(self) -> None:
         if self.is_bot:
             return
         creators = []
         limit = config["bridge.sync_dialog_limit"] or None
-        self.log.debug(f"Syncing dialogs (limit={limit}, synchronous_create={synchronous_create})")
+        self.log.debug(f"Syncing dialogs (limit={limit})")
+        dialog: Dialog
         async for dialog in self.client.iter_dialogs(limit=limit, ignore_migrated=True,
                                                      archived=False):
             entity = dialog.entity
@@ -353,11 +355,16 @@ class User(AbstractUser, BaseUser):
                 continue
             portal = po.Portal.get_by_entity(entity, receiver_id=self.tgid)
             self.portals[portal.tgid_full] = portal
-            creators.append(
-                portal.create_matrix_room(self, entity, invites=[self.mxid],
-                                          synchronous=synchronous_create))
+            if portal.mxid:
+                update_task = portal.update_matrix_room(self, entity)
+                backfill_task = portal.backfill(self, last_known_id=dialog.message.id)
+                creators.append(self.loop.create_task(update_task))
+                creators.append(self.loop.create_task(backfill_task))
+            else:
+                create_task = portal.create_matrix_room(self, entity, invites=[self.mxid])
+                creators.append(self.loop.create_task(create_task))
         self.save(portals=True)
-        await asyncio.gather(*creators, loop=self.loop)
+        await asyncio.gather(*creators)
         self.log.debug("Dialog syncing complete")
 
     def register_portal(self, portal: po.Portal) -> None: