Bump version to 0.13.0

Fixes #809

.github/workflows/python-lint.yml

@@ -9,14 +9,14 @@ jobs:
     - uses: actions/checkout@v2
     - uses: actions/setup-python@v2
       with:
-        python-version: "3.10"
+        python-version: "3.11"
     - uses: isort/isort-action@master
       with:
         sortPaths: "./mautrix_telegram"
     - uses: psf/black@stable
       with:
         src: "./mautrix_telegram"
-        version: "22.3.0"
+        version: "23.1.0"
     - name: pre-commit
       run: |
         pip install pre-commit

.gitignore

@@ -14,5 +14,8 @@ __pycache__
 /registration.yaml
 *.log*
 *.db
-*.pickle
+/*.pickle
 *.bak
+/*.session
+/*.session-journal
+/*.json

.pre-commit-config.yaml

@@ -1,6 +1,6 @@
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.1.0
+    rev: v4.4.0
     hooks:
       - id: trailing-whitespace
         exclude_types: [markdown]
@@ -8,13 +8,13 @@ repos:
       - id: check-yaml
       - id: check-added-large-files
   - repo: https://github.com/psf/black
-    rev: 22.3.0
+    rev: 23.1.0
     hooks:
       - id: black
         language_version: python3
         files: ^mautrix_telegram/.*\.pyi?$
   - repo: https://github.com/PyCQA/isort
-    rev: 5.10.1
+    rev: 5.12.0
     hooks:
       - id: isort
         files: ^mautrix_telegram/.*\.pyi?$

CHANGELOG.md

@@ -1,3 +1,55 @@
+# v0.13.0 (2023-02-26)
+
+### Added
+* Added `allow_contact_info` config option to specify whether personal names
+  and avatars for other users should be bridged.
+  * The option is only safe to enable on single-user instances, using it
+    anywhere else will cause ghost user profiles to flip back and forth between
+    personal and default ones.
+* Added config option to notify Matrix room if bridging an incoming message
+  fails.
+
+### Improved
+* Updated Docker image to Alpine 3.17.
+* Updated to Telegram API layer 152.
+* Improved handling users getting logged out.
+* Removed support for creating accounts, as Telegram only allows requesting SMS
+  login codes on the official mobile clients now.
+* Replaced moviepy with calling ffmpeg directly for generating video thumbnails.
+
+### Fixed
+* Fixed handling Telegram chat upgrades when backfilling is enabled.
+* Fixed file transfers failing if transfering the thumbnail fails.
+* Fixed bridging unnamed files with unrecognized mime types.
+* Fixed enqueueing more backfill.
+* Fixed timestamps not being saved in `telegram_file` table.
+* Fixed issues with old events being replayed if the bridge was shut down
+  uncleanly.
+
+# v0.12.2 (2022-11-26)
+
+### Added
+* Added built-in custom emoji packs to allow reacting with any standard unicode
+  emoji from Matrix (note that only premium users can use custom emojis).
+* Added infinite backfill using [MSC2716].
+  * The new system includes a backwards compatibility mechanism which uses the
+    old method of just sending events to the room. By default, MSC2716 is not
+    enabled and the legacy method will be used.
+
+### Improved
+* Redacting reactions on Matrix no longer removes the user's other reactions to
+  the same message (premium users can have up to 3 reactions per message).
+* Changes to default user permissions on Telegram are now bridged.
+* Added database index to make reaction polling more efficient
+  (thanks to [@AndrewFerr] in [#862]).
+
+### Fixed
+* Fixed provisioning API not working with URL-encoded parameters.
+
+[MSC2716]: https://github.com/matrix-org/matrix-spec-proposals/pull/2716
+[@AndrewFerr]: https://github.com/AndrewFerr
+[#862]: https://github.com/mautrix/telegram/pull/862
+
 # v0.12.1 (2022-09-26)
 
 ### Added

Dockerfile

@@ -1,6 +1,4 @@
-FROM dock.mau.dev/tulir/lottieconverter:alpine-3.16
-
-ARG TARGETARCH=amd64
+FROM dock.mau.dev/tulir/lottieconverter:alpine-3.17
 
 RUN apk add --no-cache \
       python3 py3-pip py3-setuptools py3-wheel \
@@ -15,13 +13,6 @@ RUN apk add --no-cache \
       # Indirect dependencies
       py3-idna \
       py3-rsa \
-      #moviepy
-        py3-decorator \
-        py3-tqdm \
-        py3-requests \
-        #py3-proglog \
-        #imageio
-          py3-numpy \
       #py3-telethon \ (outdated)
         # Optional for socks proxies
         py3-pysocks \

README.md

@@ -11,7 +11,7 @@ A Matrix-Telegram hybrid puppeting/relaybot bridge.
 ## Sponsors
 * [Joel Lehtonen / Zouppen](https://github.com/zouppen)
 
-### Documentation
+## Documentation
 All setup and usage instructions are located on
 [docs.mau.fi](https://docs.mau.fi/bridges/python/telegram/index.html).
 Some quick links:

dev-requirements.txt

@@ -1,3 +1,3 @@
 pre-commit>=2.10.1,<3
 isort>=5.10.1,<6
-black>=22.3,<23
+black>=23,<24

mautrix_telegram/__init__.py

@@ -1,2 +1,2 @@
-__version__ = "0.12.1"
+__version__ = "0.13.0"
 __author__ = "Tulir Asokan <tulir@maunium.net>"

mautrix_telegram/abstract_user.py

@@ -22,7 +22,7 @@ import logging
 import platform
 import time
 
-from telethon.errors import UnauthorizedError
+from telethon.errors import AuthKeyError, UnauthorizedError
 from telethon.network import (
     Connection,
     ConnectionTcpFull,
@@ -41,6 +41,7 @@ from telethon.tl.types import (
     TypeUpdate,
     UpdateChannel,
     UpdateChannelUserTyping,
+    UpdateChatDefaultBannedRights,
     UpdateChatParticipantAdmin,
     UpdateChatParticipants,
     UpdateChatUserTyping,
@@ -62,8 +63,8 @@ from telethon.tl.types import (
     UpdateShort,
     UpdateShortChatMessage,
     UpdateShortMessage,
+    UpdateUser,
     UpdateUserName,
-    UpdateUserPhoto,
     UpdateUserStatus,
     UpdateUserTyping,
     User,
@@ -74,6 +75,7 @@ from telethon.tl.types import (
 from mautrix.appservice import AppService
 from mautrix.errors import MatrixError
 from mautrix.types import PresenceState, UserID
+from mautrix.util import background_task
 from mautrix.util.logging import TraceLogger
 from mautrix.util.opt_prometheus import Counter, Histogram
 
@@ -234,18 +236,23 @@ class AbstractUser(ABC):
         )
         self.client.add_event_handler(self._update_catch)
 
+    @abstractmethod
+    async def on_signed_out(self, err: UnauthorizedError | AuthKeyError) -> None:
+        pass
+
     async def _telethon_update_error_callback(self, err: Exception) -> None:
+        if isinstance(err, (UnauthorizedError, AuthKeyError)):
+            background_task.create(self.on_signed_out(err))
+            return
         if self.config["telegram.exit_on_update_error"]:
             self.log.critical(f"Stopping due to update handling error {type(err).__name__}")
             self.bridge.manual_stop(50)
         else:
-            if isinstance(err, UnauthorizedError):
-                self.log.warning("Not recreating Telethon update loop")
-                return
-            self.log.info("Recreating Telethon update loop in 60 seconds")
+            self.log.info("Recreating Telethon connection in 60 seconds")
             await asyncio.sleep(60)
-            self.log.debug("Now recreating Telethon update loop")
-            self.client._updates_handle = self.loop.create_task(self.client._update_loop())
+            self.log.debug("Now recreating Telethon connection")
+            await self.stop()
+            await self.start()
 
     @abstractmethod
     async def update(self, update: TypeUpdate) -> bool:
@@ -319,7 +326,7 @@ class AbstractUser(ABC):
     async def _update(self, update: TypeUpdate) -> None:
         if isinstance(update, UpdateShort):
             update = update.update
-        asyncio.create_task(self._handle_entity_updates(getattr(update, "_entities", {})))
+        background_task.create(self._handle_entity_updates(getattr(update, "_entities", {})))
         if isinstance(
             update,
             (
@@ -346,9 +353,11 @@ class AbstractUser(ABC):
             await self.update_admin(update)
         elif isinstance(update, UpdateChatParticipants):
             await self.update_participants(update)
+        elif isinstance(update, UpdateChatDefaultBannedRights):
+            await self.update_default_banned_rights(update)
         elif isinstance(update, (UpdatePinnedMessages, UpdatePinnedChannelMessages)):
             await self.update_pinned_messages(update)
-        elif isinstance(update, (UpdateUserName, UpdateUserPhoto)):
+        elif isinstance(update, (UpdateUserName, UpdateUser)):
             await self.update_others_info(update)
         elif isinstance(update, UpdateReadHistoryOutbox):
             await self.update_read_receipt(update)
@@ -392,6 +401,12 @@ class AbstractUser(ABC):
         if portal and portal.mxid:
             await portal.update_power_levels(update.participants.participants)
 
+    @staticmethod
+    async def update_default_banned_rights(update: UpdateChatDefaultBannedRights) -> None:
+        portal = await po.Portal.get_by_entity(update.peer)
+        if portal and portal.mxid:
+            await portal.update_default_banned_rights(update.default_banned_rights)
+
     async def update_read_receipt(self, update: UpdateReadHistoryOutbox) -> None:
         if not isinstance(update.peer, PeerUser):
             self.log.debug("Unexpected read receipt peer: %s", update.peer)
@@ -420,6 +435,7 @@ class AbstractUser(ABC):
         if not puppet.is_real_user:
             return
 
+        self.log.debug("Handling own read receipt: %s", update)
         if isinstance(update, UpdateReadChannelInbox):
             portal = await po.Portal.get_by_tgid(TelegramID(update.channel_id))
         elif isinstance(update.peer, PeerChat):
@@ -433,6 +449,7 @@ class AbstractUser(ABC):
             return
 
         if not portal or not portal.mxid:
+            self.log.debug(f"Dropping own read receipt in unknown chat ({update.peer})")
             return
 
         tg_space = portal.tgid if portal.peer_type == "channel" else self.tgid
@@ -440,6 +457,9 @@ class AbstractUser(ABC):
             TelegramID(update.max_id), tg_space, edit_index=-1
         )
         if not message:
+            self.log.debug(
+                f"Dropping own read receipt: unknown message {update.max_id}@{tg_space}"
+            )
             return
 
         await puppet.intent.mark_read(portal.mxid, message.mxid)
@@ -486,18 +506,23 @@ class AbstractUser(ABC):
         except Exception:
             self.log.exception("Failed to handle entity updates")
 
-    async def update_others_info(self, update: UpdateUserName | UpdateUserPhoto) -> None:
+    async def update_others_info(self, update: UpdateUserName | UpdateUser) -> None:
         # TODO duplication not checked
         puppet = await pu.Puppet.get_by_tgid(TelegramID(update.user_id))
         if isinstance(update, UpdateUserName):
-            puppet.username = update.username
+            if len(update.usernames) > 1:
+                self.log.warning(
+                    "Got update with multiple usernames (%s) for %s, only saving first one",
+                    update.usernames,
+                    update.user_id,
+                )
+            puppet.username = update.usernames[0].username if update.usernames else None
             if await puppet.update_displayname(self, update):
                 await puppet.save()
                 await puppet.update_portals_meta()
-        elif isinstance(update, UpdateUserPhoto):
-            if await puppet.update_avatar(self, update.photo):
-                await puppet.save()
-                await puppet.update_portals_meta()
+        elif isinstance(update, UpdateUser):
+            info = await self.client.get_entity(puppet.peer)
+            await puppet.update_info(self, info)
         else:
             self.log.warning(f"Unexpected other user info update: {type(update)}")
 
@@ -515,9 +540,7 @@ class AbstractUser(ABC):
         self, update: UpdateMessage
     ) -> tuple[UpdateMessageContent, pu.Puppet | None, po.Portal | None]:
         if isinstance(update, UpdateShortChatMessage):
-            portal = await po.Portal.get_by_tgid(TelegramID(update.chat_id))
-            if not portal:
-                self.log.warning(f"Received message in chat with unknown type {update.chat_id}")
+            portal = await po.Portal.get_by_tgid(TelegramID(update.chat_id), peer_type="chat")
             sender = await pu.Puppet.get_by_tgid(TelegramID(update.from_id))
         elif isinstance(update, UpdateShortMessage):
             portal = await po.Portal.get_by_tgid(
@@ -541,6 +564,8 @@ class AbstractUser(ABC):
                 sender = await pu.Puppet.get_by_tgid(self.tgid)
             elif isinstance(update.from_id, (PeerUser, PeerChannel)):
                 sender = await pu.Puppet.get_by_peer(update.from_id)
+            elif isinstance(update.peer_id, PeerUser):
+                sender = await pu.Puppet.get_by_peer(update.peer_id)
             else:
                 sender = None
         else:
@@ -601,26 +626,33 @@ class AbstractUser(ABC):
             await portal.delete_telegram_user(self.tgid, sender=None)
         elif chan := getattr(update, "mau_channel", None):
             if not portal.mxid:
-                asyncio.create_task(self._delayed_create_channel(chan))
+                background_task.create(self._delayed_create_channel(chan))
             else:
                 self.log.debug("Updating channel info with data fetched by Telethon")
                 await portal.update_info(self, chan)
                 await portal.invite_to_matrix(self.mxid)
 
     async def _delayed_create_channel(self, chan: Channel) -> None:
-        self.log.debug("Waiting 5 seconds before handling UpdateChannel for non-existent portal")
+        self.log.debug(
+            f"Waiting 5 seconds before handling UpdateChannel for non-existent portal {chan.id}"
+        )
         await asyncio.sleep(5)
         portal = await po.Portal.get_by_tgid(TelegramID(chan.id))
         if portal.mxid:
             self.log.debug(
-                "Portal started existing after waiting 5 seconds, dropping UpdateChannel"
+                "Portal started existing after waiting 5 seconds, "
+                f"dropping UpdateChannel for {portal.tgid}"
             )
             return
         else:
             self.log.info(
-                "Creating Matrix room with data fetched by Telethon due to UpdateChannel"
+                f"Creating Matrix room for {portal.tgid}"
+                " with data fetched by Telethon due to UpdateChannel"
             )
-            await portal.create_matrix_room(self, chan)
+            await portal.create_matrix_room(self, chan, invites=[self.mxid])
+
+    async def _check_server_notice_edit(self, message: Message) -> None:
+        pass
 
     async def update_message(self, original_update: UpdateMessage) -> None:
         update, sender, portal = await self.get_message_details(original_update)
@@ -655,7 +687,7 @@ class AbstractUser(ABC):
 
         if isinstance(update, MessageService):
             if isinstance(update.action, MessageActionChannelMigrateFrom):
-                self.log.trace(
+                self.log.debug(
                     "Received %s in %s by %d, unregistering portal...",
                     update.action,
                     portal.tgid_log,
@@ -664,7 +696,7 @@ class AbstractUser(ABC):
                 await self.unregister_portal(update.action.chat_id, update.action.chat_id)
                 await self.register_portal(portal)
                 return
-            self.log.trace(
+            self.log.debug(
                 "Handling action %s to %s by %d",
                 update.action,
                 portal.tgid_log,
@@ -673,6 +705,8 @@ class AbstractUser(ABC):
             return await portal.handle_telegram_action(self, sender, update)
 
         if isinstance(original_update, (UpdateEditMessage, UpdateEditChannelMessage)):
+            if sender and sender.tgid == 777000:
+                await self._check_server_notice_edit(update)
             return await portal.handle_telegram_edit(self, sender, update)
         return await portal.handle_telegram_message(self, sender, update)
 

mautrix_telegram/bot.py

@@ -15,11 +15,16 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 from __future__ import annotations
 
-from typing import Awaitable, Callable, Literal
+from typing import TYPE_CHECKING, Awaitable, Callable, Literal
 import logging
 import time
 
-from telethon.errors import ChannelInvalidError, ChannelPrivateError
+from telethon.errors import (
+    AuthKeyError,
+    ChannelInvalidError,
+    ChannelPrivateError,
+    UnauthorizedError,
+)
 from telethon.tl.functions.channels import GetChannelsRequest, GetParticipantRequest
 from telethon.tl.functions.messages import GetChatsRequest, GetFullChatRequest
 from telethon.tl.patched import Message, MessageService
@@ -57,6 +62,9 @@ from .abstract_user import AbstractUser
 from .db import BotChat, Message as DBMessage
 from .types import TelegramID
 
+if TYPE_CHECKING:
+    from asyncio import Future
+
 ReplyFunc = Callable[[str], Awaitable[Message]]
 BanFunc = Callable[[RoomID, UserID, str], Awaitable[None]]
 TelegramAdminPermission = Literal[
@@ -87,6 +95,7 @@ class Bot(AbstractUser):
         tuple[int, int],
         tuple[ChatParticipantAdmin | ChatParticipantCreator | None, float],
     ]
+    _login_wait_fut: Future | None
     required_permissions: dict[str, TelegramAdminPermission] = {
         "portal": None,
         "invite": "invite_users",
@@ -141,14 +150,19 @@ class Bot(AbstractUser):
         await self.post_login()
         return self
 
+    async def on_signed_out(self, err: UnauthorizedError | AuthKeyError) -> None:
+        self.log.fatal("Relay bot got signed out, crashing bridge", exc_info=err)
+        self.bridge.manual_stop(51)
+
     async def post_login(self) -> None:
         await self.init_permissions()
         info = await self.client.get_me()
         self.tgid = TelegramID(info.id)
         self.tg_username = info.username
         self.mxid = pu.Puppet.get_mxid_from_id(self.tgid)
-        self._login_wait_fut.set_result(None)
-        self._login_wait_fut = None
+        if self._login_wait_fut:
+            self._login_wait_fut.set_result(None)
+            self._login_wait_fut = None
 
         chat_ids = [chat_id for chat_id, chat_type in self.chats.items() if chat_type == "chat"]
         response = await self.client(GetChatsRequest(chat_ids))

mautrix_telegram/commands/portal/bridge.py

@@ -21,6 +21,7 @@ import asyncio
 from telethon.tl.types import ChannelForbidden, ChatForbidden
 
 from mautrix.types import EventID, RoomID
+from mautrix.util import background_task
 
 from ... import portal as po
 from ...types import TelegramID
@@ -184,7 +185,7 @@ async def confirm_bridge(evt: CommandEvent) -> EventID | None:
         if not ok:
             return None
         elif coro:
-            asyncio.create_task(coro)
+            background_task.create(coro)
             await evt.reply("Cleaning up previous portal room...")
     elif portal.mxid:
         evt.sender.command_status = None
@@ -251,7 +252,7 @@ async def _locked_confirm_bridge(
     await portal.save()
     await portal.update_bridge_info()
 
-    asyncio.create_task(portal.update_matrix_room(user, entity, levels=levels))
+    background_task.create(portal.update_matrix_room(user, entity, levels=levels))
 
     await warn_missing_power(levels, evt)
 

mautrix_telegram/commands/telegram/auth.py

@@ -22,7 +22,6 @@ import io
 from telethon.errors import (
     AccessTokenExpiredError,
     AccessTokenInvalidError,
-    FirstNameInvalidError,
     FloodWaitError,
     PasswordHashInvalidError,
     PhoneCodeExpiredError,
@@ -31,14 +30,12 @@ from telethon.errors import (
     PhoneNumberBannedError,
     PhoneNumberFloodError,
     PhoneNumberInvalidError,
-    PhoneNumberOccupiedError,
     PhoneNumberUnoccupiedError,
     SessionPasswordNeededError,
 )
 from telethon.tl.types import User
 
 from mautrix.client import Client
-from mautrix.errors import MForbidden
 from mautrix.types import (
     EventID,
     ImageInfo,
@@ -47,6 +44,7 @@ from mautrix.types import (
     TextMessageEventContent,
     UserID,
 )
+from mautrix.util import background_task
 from mautrix.util.format_duration import format_duration as fmt_duration
 
 from ... import user as u
@@ -94,70 +92,6 @@ async def ping_bot(evt: CommandEvent) -> EventID:
     )
 
 
-@command_handler(
-    needs_auth=False,
-    management_only=True,
-    help_section=SECTION_AUTH,
-    help_args="<_phone_> <_full name_>",
-    help_text="Register to Telegram",
-)
-async def register(evt: CommandEvent) -> EventID:
-    if await evt.sender.is_logged_in():
-        return await evt.reply("You are already logged in.")
-    elif len(evt.args) < 1:
-        return await evt.reply("**Usage:** `$cmdprefix+sp register <phone> <full name>`")
-
-    phone_number = evt.args[0]
-    if len(evt.args) == 2:
-        full_name = evt.args[1], ""
-    else:
-        full_name = " ".join(evt.args[1:-1]), evt.args[-1]
-
-    await _request_code(
-        evt,
-        phone_number,
-        {
-            "next": enter_code_register,
-            "action": "Register",
-            "full_name": full_name,
-        },
-    )
-    return await evt.reply(
-        "By signing up for Telegram, you agree to "
-        "the terms of service: https://telegram.org/tos"
-    )
-
-
-async def enter_code_register(evt: CommandEvent) -> EventID:
-    if len(evt.args) == 0:
-        return await evt.reply("**Usage:** `$cmdprefix+sp <code>`")
-    try:
-        await evt.sender.ensure_started(even_if_no_session=True)
-        first_name, last_name = evt.sender.command_status["full_name"]
-        user = await evt.sender.client.sign_up(evt.args[0], first_name, last_name)
-        asyncio.create_task(evt.sender.post_login(user, first_login=True))
-        evt.sender.command_status = None
-        return await evt.reply(f"Successfully registered to Telegram.")
-    except PhoneNumberOccupiedError:
-        return await evt.reply(
-            "That phone number has already been registered. "
-            "You can log in with `$cmdprefix+sp login`."
-        )
-    except FirstNameInvalidError:
-        return await evt.reply("Invalid name. Please set a Matrix displayname before registering.")
-    except PhoneCodeExpiredError:
-        return await evt.reply(
-            "Phone code expired. Try again with `$cmdprefix+sp register <phone>`."
-        )
-    except PhoneCodeInvalidError:
-        return await evt.reply("Invalid phone code.")
-    except Exception:
-        evt.log.exception("Error sending phone code")
-        return await evt.reply(
-            "Unhandled exception while sending code. Check console for more details."
-        )
-
-
 @command_handler(
     needs_auth=False,
     management_only=True,
@@ -317,7 +251,7 @@ async def _request_code(
     except PhoneNumberUnoccupiedError:
         return await evt.reply(
             "That phone number has not been registered. "
-            "Please register with `$cmdprefix+sp register <phone>`."
+            "Please sign up to Telegram using an official mobile client first."
         )
     except PhoneNumberInvalidError:
         return await evt.reply("That phone number is not valid.")
@@ -432,7 +366,7 @@ async def _finish_sign_in(evt: CommandEvent, user: User, login_as: u.User = None
             f"[{existing_user.displayname}] (https://matrix.to/#/{existing_user.mxid})"
             " was logged out from the account."
         )
-    asyncio.create_task(login_as.post_login(user, first_login=True))
+    background_task.create(login_as.post_login(user, first_login=True))
     evt.sender.command_status = None
     name = f"@{user.username}" if user.username else f"+{user.phone}"
     if login_as != evt.sender:

mautrix_telegram/commands/telegram/misc.py

@@ -18,6 +18,7 @@ from __future__ import annotations
 from typing import cast
 import base64
 import codecs
+import math
 import re
 
 from aiohttp import ClientSession, InvalidURL
@@ -133,15 +134,16 @@ async def search(evt: CommandEvent) -> EventID:
 
 @command_handler(
     help_section=SECTION_CREATING_PORTALS,
-    help_args="<_identifier_>",
-    help_text="Open a private chat with the given Telegram user. The identifier is "
-    "either the internal user ID, the username or the phone number. "
-    "**N.B.** The phone numbers you start chats with must already be in "
-    "your contacts.",
+    help_args="<_username_>",
+    help_text=(
+        "Open a private chat with the given Telegram user. You can also use a "
+        "phone number instead of username, but you must have the number in "
+        "your Telegram contacts for that to work."
+    ),
 )
 async def pm(evt: CommandEvent) -> EventID:
     if len(evt.args) == 0:
-        return await evt.reply("**Usage:** `$cmdprefix+sp pm <user identifier>`")
+        return await evt.reply("**Usage:** `$cmdprefix+sp pm <username>`")
 
     try:
         id = "".join(evt.args).translate({ord(c): None for c in "+()- "})
@@ -427,6 +429,9 @@ async def backfill(evt: CommandEvent) -> None:
     if not evt.is_portal:
         await evt.reply("You can only use backfill in portal rooms")
         return
+    elif not evt.config["bridge.backfill.enable"]:
+        await evt.reply("Backfilling is disabled in the bridge config")
+        return
     try:
         limit = int(evt.args[0])
     except (ValueError, IndexError):
@@ -435,16 +440,14 @@ async def backfill(evt: CommandEvent) -> None:
     if not evt.config["bridge.backfill.normal_groups"] and portal.peer_type == "chat":
         await evt.reply("Backfilling normal groups is disabled in the bridge config")
         return
-    try:
-        await portal.backfill(evt.sender, limit=limit)
-    except TakeoutInitDelayError:
-        msg = (
-            "Please accept the data export request from a mobile device, "
-            "then re-run the backfill command."
-        )
-        if portal.peer_type == "user":
-            from mautrix.appservice import IntentAPI
-
-            await portal.main_intent.send_notice(evt.room_id, msg)
-        else:
-            await evt.reply(msg)
+    if portal.backfill_msc2716:
+        messages_per_batch = evt.config["bridge.backfill.incremental.messages_per_batch"]
+        batches = math.ceil(limit / messages_per_batch)
+        rounded = ""
+        if batches * messages_per_batch != limit:
+            rounded = f" (rounded message limit to {batches}*{messages_per_batch})"
+        await portal.enqueue_backfill(evt.sender, priority=0, max_batches=batches)
+        await evt.reply(f"Backfill queued{rounded}")
+    else:
+        output = await portal.forward_backfill(evt.sender, initial=False, override_limit=limit)
+        await evt.reply(output)

mautrix_telegram/config.py

@@ -101,6 +101,7 @@ class Config(BaseBridgeConfig):
         copy("bridge.displayname_preference")
         copy("bridge.displayname_max_length")
         copy("bridge.allow_avatar_remove")
+        copy("bridge.allow_contact_info")
 
         copy("bridge.max_initial_member_sync")
         copy("bridge.max_member_count")
@@ -113,6 +114,7 @@ class Config(BaseBridgeConfig):
         else:
             copy("bridge.sync_update_limit")
             copy("bridge.sync_create_limit")
+        copy("bridge.sync_deferred_create_all")
         copy("bridge.sync_direct_chats")
         copy("bridge.max_telegram_delete")
         copy("bridge.sync_matrix_state")
@@ -136,6 +138,7 @@ class Config(BaseBridgeConfig):
         copy("bridge.image_as_file_pixels")
         copy("bridge.parallel_file_transfer")
         copy("bridge.federate_rooms")
+        copy("bridge.always_custom_emoji_reaction")
         copy("bridge.animated_sticker.target")
         copy("bridge.animated_sticker.convert_from_webm")
         copy("bridge.animated_sticker.args.width")
@@ -148,6 +151,7 @@ class Config(BaseBridgeConfig):
         copy("bridge.private_chat_portal_meta")
         copy("bridge.delivery_receipts")
         copy("bridge.delivery_error_reports")
+        copy("bridge.incoming_bridge_error_reports")
         copy("bridge.message_status_events")
         copy("bridge.resend_bridge_info")
         copy("bridge.mute_bridging")
@@ -157,12 +161,19 @@ class Config(BaseBridgeConfig):
         copy("bridge.bridge_matrix_leave")
         copy("bridge.kick_on_logout")
         copy("bridge.always_read_joined_telegram_notice")
-        copy("bridge.backfill.invite_own_puppet")
-        copy("bridge.backfill.takeout_limit")
-        copy("bridge.backfill.initial_limit")
-        copy("bridge.backfill.missed_limit")
-        copy("bridge.backfill.disable_notifications")
+        copy("bridge.backfill.enable")
+        copy("bridge.backfill.msc2716")
+        copy("bridge.backfill.double_puppet_backfill")
         copy("bridge.backfill.normal_groups")
+        copy("bridge.backfill.unread_hours_threshold")
+        copy("bridge.backfill.forward.initial_limit")
+        copy("bridge.backfill.forward.sync_limit")
+        copy("bridge.backfill.incremental.messages_per_batch")
+        copy("bridge.backfill.incremental.post_batch_delay")
+        copy("bridge.backfill.incremental.max_batches.user")
+        copy("bridge.backfill.incremental.max_batches.normal_group")
+        copy("bridge.backfill.incremental.max_batches.supergroup")
+        copy("bridge.backfill.incremental.max_batches.channel")
 
         copy("bridge.initial_power_level_overrides.group")
         copy("bridge.initial_power_level_overrides.user")

mautrix_telegram/db/__init__.py

@@ -15,6 +15,7 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 from mautrix.util.async_db import Database
 
+from .backfill_queue import Backfill, BackfillType
 from .bot_chat import BotChat
 from .disappearing_message import DisappearingMessage
 from .message import Message
@@ -38,6 +39,7 @@ def init(db: Database) -> None:
         BotChat,
         PgSession,
         DisappearingMessage,
+        Backfill,
     ):
         table.db = db
 
@@ -54,4 +56,5 @@ __all__ = [
     "BotChat",
     "PgSession",
     "DisappearingMessage",
+    "Backfill",
 ]

mautrix_telegram/db/backfill_queue.py

@@ -0,0 +1,235 @@
+# mautrix-telegram - A Matrix-Telegram puppeting bridge
+# Copyright (C) 2022 Tulir Asokan, Sumner Evans
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, ClassVar
+from datetime import datetime, timedelta
+from enum import Enum
+import json
+
+from asyncpg import Record
+from attr import dataclass
+
+from mautrix.types import UserID
+from mautrix.util.async_db import Connection, Database
+
+from ..types import TelegramID
+
+fake_db = Database.create("") if TYPE_CHECKING else None
+
+
+class BackfillType(Enum):
+    HISTORICAL = "historical"
+    SYNC_DIALOG = "sync_dialog"
+
+
+@dataclass
+class Backfill:
+    db: ClassVar[Database] = fake_db
+
+    queue_id: int | None
+    user_mxid: UserID
+    priority: int
+    type: BackfillType
+    portal_tgid: TelegramID
+    portal_tg_receiver: TelegramID
+    anchor_msg_id: TelegramID | None
+    extra_data: dict[str, Any]
+    messages_per_batch: int
+    post_batch_delay: int
+    max_batches: int
+    dispatch_time: datetime | None
+    completed_at: datetime | None
+    cooldown_timeout: datetime | None
+
+    @staticmethod
+    def new(
+        user_mxid: UserID,
+        priority: int,
+        type: BackfillType,
+        portal_tgid: TelegramID,
+        portal_tg_receiver: TelegramID,
+        messages_per_batch: int,
+        anchor_msg_id: TelegramID | None = None,
+        extra_data: dict[str, Any] | None = None,
+        post_batch_delay: int = 0,
+        max_batches: int = -1,
+    ) -> "Backfill":
+        return Backfill(
+            queue_id=None,
+            user_mxid=user_mxid,
+            priority=priority,
+            type=type,
+            portal_tgid=portal_tgid,
+            portal_tg_receiver=portal_tg_receiver,
+            anchor_msg_id=anchor_msg_id,
+            extra_data=extra_data or {},
+            messages_per_batch=messages_per_batch,
+            post_batch_delay=post_batch_delay,
+            max_batches=max_batches,
+            dispatch_time=None,
+            completed_at=None,
+            cooldown_timeout=None,
+        )
+
+    @classmethod
+    def _from_row(cls, row: Record | None) -> Backfill | None:
+        if row is None:
+            return None
+        data = {**row}
+        type = BackfillType(data.pop("type"))
+        extra_data = json.loads(data.pop("extra_data", None) or "{}")
+        return cls(**data, type=type, extra_data=extra_data)
+
+    columns = [
+        "user_mxid",
+        "priority",
+        "type",
+        "portal_tgid",
+        "portal_tg_receiver",
+        "anchor_msg_id",
+        "extra_data",
+        "messages_per_batch",
+        "post_batch_delay",
+        "max_batches",
+        "dispatch_time",
+        "completed_at",
+        "cooldown_timeout",
+    ]
+    columns_str = ",".join(columns)
+
+    @classmethod
+    async def get_next(cls, user_mxid: UserID) -> Backfill | None:
+        q = f"""
+        SELECT queue_id, {cls.columns_str}
+        FROM backfill_queue
+        WHERE user_mxid=$1
+            AND (
+                dispatch_time IS NULL
+                OR (
+                    dispatch_time < $2
+                    AND completed_at IS NULL
+                )
+            )
+            AND (
+                cooldown_timeout IS NULL
+                OR cooldown_timeout < current_timestamp
+            )
+        ORDER BY priority, queue_id
+        LIMIT 1
+        """
+        return cls._from_row(
+            await cls.db.fetchrow(q, user_mxid, datetime.now() - timedelta(minutes=15))
+        )
+
+    @classmethod
+    async def delete_existing(
+        cls,
+        user_mxid: UserID,
+        portal_tgid: int,
+        portal_tg_receiver: int,
+        type: BackfillType,
+    ) -> Backfill | None:
+        q = f"""
+        WITH deleted_entries AS (
+            DELETE FROM backfill_queue
+            WHERE user_mxid=$1
+              AND portal_tgid=$2
+              AND portal_tg_receiver=$3
+              AND type=$4
+              AND dispatch_time IS NULL
+              AND completed_at IS NULL
+            RETURNING 1
+        )
+        WITH dispatched_entries AS (
+            SELECT 1 FROM backfill_queue
+            WHERE user_mxid=$1
+              AND portal_tgid=$2
+              AND portal_tg_receiver=$3
+              AND type=$4
+              AND dispatch_time IS NOT NULL
+              AND completed_at IS NULL
+        )
+        """
+        return cls._from_row(
+            await cls.db.fetchrow(q, user_mxid, portal_tgid, portal_tg_receiver, type.value)
+        )
+
+    @classmethod
+    async def delete_all(cls, user_mxid: UserID, conn: Connection | None = None) -> None:
+        await (conn or cls.db).execute("DELETE FROM backfill_queue WHERE user_mxid=$1", user_mxid)
+
+    @classmethod
+    async def delete_for_portal(cls, tgid: int, tg_receiver: int) -> None:
+        q = "DELETE FROM backfill_queue WHERE portal_tgid=$1 AND portal_tg_receiver=$2"
+        await cls.db.execute(q, tgid, tg_receiver)
+
+    async def insert(self) -> list[Backfill]:
+        delete_q = f"""
+        DELETE FROM backfill_queue
+        WHERE user_mxid=$1
+          AND portal_tgid=$2
+          AND portal_tg_receiver=$3
+          AND type=$4
+          AND dispatch_time IS NULL
+          AND completed_at IS NULL
+        RETURNING queue_id, {self.columns_str}
+        """
+        q = f"""
+        INSERT INTO backfill_queue ({self.columns_str})
+        VALUES ({','.join(f'${i+1}' for i in range(len(self.columns)))})
+        RETURNING queue_id
+        """
+        async with self.db.acquire() as conn, conn.transaction():
+            deleted_rows = await conn.fetch(
+                delete_q,
+                self.user_mxid,
+                self.portal_tgid,
+                self.portal_tg_receiver,
+                self.type.value,
+            )
+            self.queue_id = await conn.fetchval(
+                q,
+                self.user_mxid,
+                self.priority,
+                self.type.value,
+                self.portal_tgid,
+                self.portal_tg_receiver,
+                self.anchor_msg_id,
+                json.dumps(self.extra_data) if self.extra_data else None,
+                self.messages_per_batch,
+                self.post_batch_delay,
+                self.max_batches,
+                self.dispatch_time,
+                self.completed_at,
+                self.cooldown_timeout,
+            )
+        return [self._from_row(row) for row in deleted_rows]
+
+    async def mark_dispatched(self) -> None:
+        q = "UPDATE backfill_queue SET dispatch_time=$1 WHERE queue_id=$2"
+        await self.db.execute(q, datetime.now(), self.queue_id)
+
+    async def mark_done(self) -> None:
+        q = "UPDATE backfill_queue SET completed_at=$1 WHERE queue_id=$2"
+        await self.db.execute(q, datetime.now(), self.queue_id)
+
+    async def set_cooldown_timeout(self, timeout: int) -> None:
+        """
+        Set the backfill request to cooldown for ``timeout`` seconds.
+        """
+        q = "UPDATE backfill_queue SET cooldown_timeout=$1 WHERE queue_id=$2"
+        await self.db.execute(q, datetime.now() + timedelta(seconds=timeout), self.queue_id)

mautrix_telegram/db/message.py

@@ -19,6 +19,7 @@ from typing import TYPE_CHECKING, ClassVar
 
 from asyncpg import Record
 from attr import dataclass
+import attr
 
 from mautrix.types import EventID, RoomID, UserID
 from mautrix.util.async_db import Database, Scheme
@@ -122,6 +123,14 @@ class Message:
         )
         return cls._from_row(await cls.db.fetchrow(q, mx_room, tg_space))
 
+    @classmethod
+    async def find_first(cls, mx_room: RoomID, tg_space: TelegramID) -> Message | None:
+        q = (
+            f"SELECT {cls.columns} FROM message WHERE mx_room=$1 AND tg_space=$2 "
+            f"ORDER BY tgid ASC LIMIT 1"
+        )
+        return cls._from_row(await cls.db.fetchrow(q, mx_room, tg_space))
+
     @classmethod
     async def delete_all(cls, mx_room: RoomID) -> None:
         await cls.db.execute("DELETE FROM message WHERE mx_room=$1", mx_room)
@@ -173,6 +182,23 @@ class Message:
         q = "DELETE FROM message WHERE mxid=$1 AND mx_room=$2"
         await cls.db.execute(q, temp_mxid, mx_room)
 
+    @classmethod
+    async def bulk_insert(cls, messages: list[Message]) -> None:
+        columns = cls.columns.split(", ")
+        records = [attr.astuple(message) for message in messages]
+        async with cls.db.acquire() as conn, conn.transaction():
+            if cls.db.scheme == Scheme.POSTGRES:
+                await conn.copy_records_to_table("message", records=records, columns=columns)
+            else:
+                await conn.executemany(cls._insert_query, records)
+
+    _insert_query: ClassVar[
+        str
+    ] = """
+        INSERT INTO message (mxid, mx_room, tgid, tg_space, edit_index, redacted, content_hash, sender_mxid, sender)
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+    """
+
     @property
     def _values(self):
         return (
@@ -188,13 +214,7 @@ class Message:
         )
 
     async def insert(self) -> None:
-        q = """
-            INSERT INTO message (
-                mxid, mx_room, tgid, tg_space, edit_index, redacted, content_hash,
-                sender_mxid, sender
-            ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
-        """
-        await self.db.execute(q, *self._values)
+        await self.db.execute(self._insert_query, *self._values)
 
     async def delete(self) -> None:
         q = "DELETE FROM message WHERE mxid=$1 AND mx_room=$2 AND tg_space=$3"

mautrix_telegram/db/portal.py

@@ -167,7 +167,10 @@ class Portal:
             "UPDATE portal SET tgid=$1, tg_receiver=$1, peer_type=$2 "
             "WHERE tgid=$3 AND tg_receiver=$3"
         )
-        await self.db.execute(q, id, peer_type, self.tgid)
+        clear_queue = "DELETE FROM backfill_queue WHERE portal_tgid=$1 AND portal_tg_receiver=$2"
+        async with self.db.acquire() as conn, conn.transaction():
+            await conn.execute(clear_queue, self.tgid, self.tg_receiver)
+            await conn.execute(q, id, peer_type, self.tgid)
         self.tgid = id
         self.tg_receiver = id
         self.peer_type = peer_type

mautrix_telegram/db/telegram_file.py

@@ -92,9 +92,9 @@ class TelegramFile:
 
     async def insert(self) -> None:
         q = (
-            "INSERT INTO telegram_file (id, mxc, mime_type, was_converted, size, width, height, "
-            "                           thumbnail, decryption_info) "
-            "VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)"
+            "INSERT INTO telegram_file (id, mxc, mime_type, was_converted, timestamp,"
+            "                           size, width, height, thumbnail, decryption_info) "
+            "VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)"
         )
         await self.db.execute(
             q,
@@ -102,6 +102,7 @@ class TelegramFile:
             self.mxc,
             self.mime_type,
             self.was_converted,
+            self.timestamp,
             self.size,
             self.width,
             self.height,

mautrix_telegram/db/telethon_session.py

@@ -123,19 +123,55 @@ class PgSession(MemorySession):
         date = datetime.datetime.utcfromtimestamp(row["date"])
         return updates.State(row["pts"], row["qts"], date, row["seq"], row["unread_count"])
 
+    _set_update_state_q = """
+    INSERT INTO telethon_update_state (session_id, entity_id, pts, qts, date, seq, unread_count)
+    VALUES ($1, $2, $3, $4, $5, $6, $7)
+    ON CONFLICT (session_id, entity_id) DO UPDATE SET
+        pts=excluded.pts, qts=excluded.qts, date=excluded.date, seq=excluded.seq,
+        unread_count=excluded.unread_count
+    """
+
     async def set_update_state(self, entity_id: int, row: updates.State) -> None:
-        q = """
-        INSERT INTO telethon_update_state(session_id, entity_id, pts, qts, date, seq, unread_count)
-        VALUES ($1, $2, $3, $4, $5, $6, $7)
-        ON CONFLICT (session_id, entity_id) DO UPDATE SET
-            pts=excluded.pts, qts=excluded.qts, date=excluded.date, seq=excluded.seq,
-            unread_count=excluded.unread_count
-        """
+        q = self._set_update_state_q
         ts = row.date.timestamp()
         await self.db.execute(
             q, self.session_id, entity_id, row.pts, row.qts, ts, row.seq, row.unread_count
         )
 
+    async def set_update_states(self, rows: list[tuple[int, updates.State]]) -> None:
+        rows = [
+            (
+                self.session_id,
+                entity_id,
+                row.pts,
+                row.qts,
+                row.date.timestamp(),
+                row.seq,
+                row.unread_count,
+            )
+            for entity_id, row in rows
+        ]
+        if self.db.scheme == Scheme.POSTGRES:
+            q = """
+            INSERT INTO telethon_update_state (
+                session_id, entity_id, pts, qts, date, seq, unread_count
+            )
+            VALUES (
+                $1,
+                unnest($2::bigint[]), unnest($3::bigint[]), unnest($4::bigint[]),
+                unnest($5::bigint[]), unnest($6::bigint[]), unnest($7::integer[])
+            )
+            ON CONFLICT (session_id, entity_id) DO UPDATE SET
+                pts=excluded.pts, qts=excluded.qts, date=excluded.date, seq=excluded.seq,
+                unread_count=excluded.unread_count
+            """
+            _, entity_ids, ptses, qtses, timestamps, seqs, unread_counts = zip(*rows)
+            await self.db.execute(
+                q, self.session_id, entity_ids, ptses, qtses, timestamps, seqs, unread_counts
+            )
+        else:
+            await self.db.executemany(self._set_update_state_q, rows)
+
     async def delete_update_state(self, entity_id: int) -> None:
         q = "DELETE FROM telethon_update_state WHERE session_id=$1 AND entity_id=$2"
         await self.db.execute(q, self.session_id, entity_id)

mautrix_telegram/db/upgrade/__init__.py

@@ -16,4 +16,8 @@ from . import (
     v11_backfill_queue,
     v12_message_sender,
     v13_multiple_reactions,
+    v14_puppet_custom_mxid_index,
+    v15_backfill_anchor_id,
+    v16_backfill_type,
+    v17_message_find_recent,
 )

mautrix_telegram/db/upgrade/v00_latest_revision.py

@@ -15,7 +15,7 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 from mautrix.util.async_db import Connection, Scheme
 
-latest_version = 13
+latest_version = 17
 
 
 async def create_latest_tables(conn: Connection, scheme: Scheme) -> int:
@@ -73,6 +73,7 @@ async def create_latest_tables(conn: Connection, scheme: Scheme) -> int:
             UNIQUE (mxid, mx_room, tg_space)
         )"""
     )
+    await conn.execute("CREATE INDEX message_mx_room_and_tgid_idx ON message(mx_room, tgid DESC)")
     await conn.execute(
         """CREATE TABLE reaction (
             mxid      TEXT NOT NULL,
@@ -123,6 +124,7 @@ async def create_latest_tables(conn: Connection, scheme: Scheme) -> int:
         )"""
     )
     await conn.execute("CREATE INDEX puppet_username_idx ON puppet(LOWER(username))")
+    await conn.execute("CREATE INDEX puppet_custom_mxid_idx ON puppet(custom_mxid)")
     await conn.execute(
         """CREATE TABLE telegram_file (
             id              TEXT PRIMARY KEY,
@@ -218,8 +220,11 @@ async def create_latest_tables(conn: Connection, scheme: Scheme) -> int:
             queue_id            INTEGER PRIMARY KEY {gen},
             user_mxid           TEXT,
             priority            INTEGER NOT NULL,
+            type                TEXT NOT NULL,
             portal_tgid         BIGINT,
             portal_tg_receiver  BIGINT,
+            anchor_msg_id       BIGINT,
+            extra_data          jsonb,
             messages_per_batch  INTEGER NOT NULL,
             post_batch_delay    INTEGER NOT NULL,
             max_batches         INTEGER NOT NULL,

mautrix_telegram/db/upgrade/v14_puppet_custom_mxid_index.py

@@ -0,0 +1,23 @@
+# mautrix-telegram - A Matrix-Telegram puppeting bridge
+# Copyright (C) 2022 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from mautrix.util.async_db import Connection
+
+from . import upgrade_table
+
+
+@upgrade_table.register(description="Add index to puppet custom_mxid column")
+async def upgrade_v14(conn: Connection) -> None:
+    await conn.execute("CREATE INDEX IF NOT EXISTS puppet_custom_mxid_idx ON puppet(custom_mxid)")

mautrix_telegram/db/upgrade/v15_backfill_anchor_id.py

@@ -0,0 +1,23 @@
+# mautrix-telegram - A Matrix-Telegram puppeting bridge
+# Copyright (C) 2022 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from mautrix.util.async_db import Connection
+
+from . import upgrade_table
+
+
+@upgrade_table.register(description="Store lowest message ID in backfill queue")
+async def upgrade_v15(conn: Connection) -> None:
+    await conn.execute("ALTER TABLE backfill_queue ADD COLUMN anchor_msg_id BIGINT")

mautrix_telegram/db/upgrade/v16_backfill_type.py

@@ -0,0 +1,28 @@
+# mautrix-telegram - A Matrix-Telegram puppeting bridge
+# Copyright (C) 2022 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from mautrix.util.async_db import Connection, Scheme
+
+from . import upgrade_table
+
+
+@upgrade_table.register(description="Add type for backfill queue items")
+async def upgrade_v16(conn: Connection, scheme: Scheme) -> None:
+    await conn.execute(
+        "ALTER TABLE backfill_queue ADD COLUMN type TEXT NOT NULL DEFAULT 'historical'"
+    )
+    await conn.execute("ALTER TABLE backfill_queue ADD COLUMN extra_data jsonb")
+    if scheme != Scheme.SQLITE:
+        await conn.execute("ALTER TABLE backfill_queue ALTER COLUMN type DROP DEFAULT")

mautrix_telegram/db/upgrade/v17_message_find_recent.py

@@ -0,0 +1,25 @@
+# mautrix-telegram - A Matrix-Telegram puppeting bridge
+# Copyright (C) 2022 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from mautrix.util.async_db import Connection
+
+from . import upgrade_table
+
+
+@upgrade_table.register(description="Add index for Message.find_recent")
+async def upgrade_v17(conn: Connection) -> None:
+    await conn.execute(
+        "CREATE INDEX IF NOT EXISTS message_mx_room_and_tgid_idx ON message(mx_room, tgid DESC)"
+    )

mautrix_telegram/db/user.py

@@ -21,9 +21,10 @@ from asyncpg import Record
 from attr import dataclass
 
 from mautrix.types import UserID
-from mautrix.util.async_db import Database, Scheme
+from mautrix.util.async_db import Connection, Database, Scheme
 
 from ..types import TelegramID
+from .backfill_queue import Backfill
 
 fake_db = Database.create("") if TYPE_CHECKING else None
 
@@ -73,6 +74,20 @@ class User:
     async def delete(self) -> None:
         await self.db.execute('DELETE FROM "user" WHERE mxid=$1', self.mxid)
 
+    async def remove_tgid(self) -> None:
+        async with self.db.acquire() as conn, conn.transaction():
+            if self.tgid:
+                await conn.execute('DELETE FROM contact WHERE "user"=$1', self.tgid)
+                await conn.execute('DELETE FROM user_portal WHERE "user"=$1', self.tgid)
+            await Backfill.delete_all(self.mxid, conn=conn)
+            self.tgid = None
+            self.tg_username = None
+            self.tg_phone = None
+            self.is_bot = False
+            self.is_premium = False
+            self.saved_contacts = 0
+            await self.save(conn=conn)
+
     @property
     def _values(self):
         return (
@@ -85,13 +100,13 @@ class User:
             self.saved_contacts,
         )
 
-    async def save(self) -> None:
+    async def save(self, conn: Connection | None = None) -> None:
         q = """
         UPDATE "user" SET tgid=$2, tg_username=$3, tg_phone=$4, is_bot=$5, is_premium=$6,
                           saved_contacts=$7
         WHERE mxid=$1
         """
-        await self.db.execute(q, *self._values)
+        await (conn or self.db).execute(q, *self._values)
 
     async def insert(self) -> None:
         q = """

mautrix_telegram/example-config.yaml

@@ -145,6 +145,9 @@ bridge:
     # as there's no way to determine whether an avatar is removed or just hidden from some users. If
     # you're on a single-user instance, this should be safe to enable.
     allow_avatar_remove: false
+    # Should contact names and profile pictures be allowed?
+    # This is only safe to enable on single-user instances.
+    allow_contact_info: false
 
     # Maximum number of members to sync per portal when starting up. Other members will be
     # synced when they send messages. The maximum is 10000, after which the Telegram server
@@ -169,7 +172,10 @@ bridge:
     sync_update_limit: 0
     # Number of most recently active dialogs to create portals for when syncing chats.
     # Set to 0 to remove limit.
-    sync_create_limit: 30
+    sync_create_limit: 15
+    # Should all chats be scheduled to be created later?
+    # This is best used in combination with MSC2716 infinite backfill.
+    sync_deferred_create_all: false
     # Whether or not to sync and create portals for direct chats at startup.
     sync_direct_chats: false
     # The maximum number of simultaneous Telegram deletions to handle.
@@ -224,6 +230,9 @@ bridge:
     # Whether or not created rooms should have federation enabled.
     # If false, created portal rooms will never be federated.
     federate_rooms: true
+    # Should the bridge send all unicode reactions as custom emoji reactions to Telegram?
+    # By default, the bridge only uses custom emojis for unicode emojis that aren't allowed in reactions.
+    always_custom_emoji_reaction: false
     # Settings for converting animated stickers.
     animated_sticker:
         # Format to which animated stickers should be converted.
@@ -308,6 +317,8 @@ bridge:
     delivery_receipts: false
     # Whether or not delivery errors should be reported as messages in the Matrix room.
     delivery_error_reports: false
+    # Should errors in incoming message handling send a message to the Matrix room?
+    incoming_bridge_error_reports: false
     # Whether the bridge should send the message status as a custom com.beeper.message_send_status event.
     message_status_events: false
     # Set this to true to tell the bridge to re-send m.bridge events to all rooms on the next run.
@@ -334,34 +345,60 @@ bridge:
     create_group_on_invite: true
     # Settings for backfilling messages from Telegram.
     backfill:
-        # Whether or not the Telegram ghosts of logged in Matrix users should be
-        # invited to private chats when backfilling history from Telegram. This is
-        # usually needed to prevent rate limits and to allow timestamp massaging.
-        invite_own_puppet: true
-        # Maximum number of messages to backfill without using a takeout.
-        # The first time a takeout is used, the user has to manually approve it from a different
-        # device. If initial_limit or missed_limit are higher than this value, the bridge will ask
-        # the user to accept the takeout after logging in before syncing any chats.
-        takeout_limit: 100
-        # Maximum number of messages to backfill initially.
-        # Set to 0 to disable backfilling when creating portal, or -1 to disable the limit.
+        # Allow backfilling at all?
+        enable: true
+        # Use MSC2716 for backfilling?
         #
-        # N.B. Initial backfill will only start after member sync. Make sure your
-        #      max_initial_member_sync is set to a low enough value so it doesn't take forever.
-        initial_limit: 0
-        # Maximum number of messages to backfill if messages were missed while the bridge was
-        # disconnected. Note that this only works for logged in users and only if the chat isn't
-        # older than sync_update_limit
-        # Set to 0 to disable backfilling missed messages.
-        missed_limit: 50
-        # If using double puppeting, should notifications be disabled
-        # while the initial backfill is in progress?
-        disable_notifications: false
+        # This requires a server with MSC2716 support, which is currently an experimental feature in Synapse.
+        # It can be enabled by setting experimental_features -> msc2716_enabled to true in homeserver.yaml.
+        msc2716: false
+        # Use double puppets for backfilling?
+        #
+        # If using MSC2716, the double puppets must be in the appservice's user ID namespace
+        # (because the bridge can't use the double puppet access token with batch sending).
+        #
+        # Even without MSC2716, bridging old messages with correct timestamps requires the double
+        # puppets to be in an appservice namespace, or the server to be modified to allow
+        # overriding timestamps anyway.
+        double_puppet_backfill: false
         # Whether or not to enable backfilling in normal groups.
         # Normal groups have numerous technical problems in Telegram, and backfilling normal groups
         # will likely cause problems if there are multiple Matrix users in the group.
         normal_groups: false
 
+        # If a backfilled chat is older than this number of hours, mark it as read even if it's unread on Telegram.
+        # Set to -1 to let any chat be unread.
+        unread_hours_threshold: 720
+
+        # Forward backfilling limits. These apply to both MSC2716 and legacy backfill.
+        #
+        # Using a negative initial limit is not recommended, as it would try to backfill everything in a single batch.
+        # MSC2716 and the incremental settings are meant for backfilling everything incrementally rather than at once.
+        forward:
+            # Number of messages to backfill immediately after creating a portal.
+            initial_limit: 10
+            # Number of messages to backfill when syncing chats.
+            sync_limit: 100
+
+        # Settings for incremental backfill of history. These only apply when using MSC2716.
+        incremental:
+            # Maximum number of messages to backfill per batch.
+            messages_per_batch: 100
+            # The number of seconds to wait after backfilling the batch of messages.
+            post_batch_delay: 20
+            # The maximum number of batches to backfill per portal, split by the chat type.
+            # If set to -1, all messages in the chat will eventually be backfilled.
+            max_batches:
+                # Direct chats
+                user: -1
+                # Normal groups. Note that the normal_groups option above must be enabled
+                # for these to be backfilled.
+                normal_group: -1
+                # Supergroups
+                supergroup: 10
+                # Broadcast channels
+                channel: -1
+
     # Overrides for base power levels.
     initial_power_level_overrides:
         user: {}

mautrix_telegram/formatter/from_matrix/__init__.py

@@ -59,7 +59,7 @@ async def matrix_to_telegram(
     if html is not None:
         return await _matrix_html_to_telegram(client, html)
     elif text is not None:
-        return _matrix_text_to_telegram(text), []
+        return _matrix_text_to_telegram(text)
     else:
         raise ValueError("text or html must be provided to convert formatting")
 
@@ -98,8 +98,13 @@ def _cut_long_message(
     return message, entities
 
 
-def _matrix_text_to_telegram(text: str) -> str:
+def _matrix_text_to_telegram(text: str) -> tuple[str, list[TypeMessageEntity]]:
     text = command_regex.sub(r"/\1", text)
     text = text.replace("\t", " " * 4)
     text = not_command_regex.sub(r"\1", text)
-    return text
+    entities = []
+    surrogated_text = add_surrogate(text)
+    if len(surrogated_text) > MAX_LENGTH:
+        surrogated_text, entities = _cut_long_message(surrogated_text, entities)
+        text = del_surrogate(surrogated_text)
+    return text, entities

mautrix_telegram/formatter/from_telegram.py

@@ -23,6 +23,9 @@ from telethon.errors import RPCError
 from telethon.helpers import add_surrogate, del_surrogate
 from telethon.tl.custom import Message
 from telethon.tl.types import (
+    Channel,
+    InputPeerChannelFromMessage,
+    InputPeerUserFromMessage,
     MessageEntityBlockquote,
     MessageEntityBold,
     MessageEntityBotCommand,
@@ -47,21 +50,47 @@ from telethon.tl.types import (
     PeerUser,
     SponsoredMessage,
     TypeMessageEntity,
+    User,
 )
 
 from mautrix.types import Format, MessageType, TextMessageEventContent
 
 from .. import abstract_user as au, portal as po, puppet as pu, user as u
 from ..db import Message as DBMessage, TelegramFile as DBTelegramFile
+from ..tgclient import MautrixTelegramClient
 from ..types import TelegramID
-from ..util.file_transfer import transfer_custom_emojis_to_matrix
+from ..util.file_transfer import UnicodeCustomEmoji, transfer_custom_emojis_to_matrix
 
 log: logging.Logger = logging.getLogger("mau.fmt.tg")
 
 
+async def _get_fwd_entity(client: MautrixTelegramClient, evt: Message) -> Channel | User | None:
+    try:
+        return await client.get_entity(evt.fwd_from.from_id)
+    except (ValueError, RPCError) as e:
+        try:
+            input_peer = await client.get_input_entity(evt.peer_id)
+            if isinstance(evt.fwd_from.from_id, PeerUser):
+                return await client.get_entity(
+                    InputPeerUserFromMessage(
+                        peer=input_peer, msg_id=evt.id, user_id=evt.fwd_from.from_id.user_id
+                    )
+                )
+            elif isinstance(evt.fwd_from.from_id, PeerChannel):
+                return await client.get_entity(
+                    InputPeerChannelFromMessage(
+                        peer=input_peer, msg_id=evt.id, channel_id=evt.fwd_from.from_id.channel_id
+                    )
+                )
+        except (ValueError, RPCError) as e:
+            pass
+        return None
+
+
 async def _add_forward_header(
-    source: au.AbstractUser, content: TextMessageEventContent, fwd_from: MessageFwdHeader
+    client: MautrixTelegramClient, content: TextMessageEventContent, evt: Message
 ) -> None:
+    fwd_from = evt.fwd_from
     fwd_from_html, fwd_from_text = None, None
     if isinstance(fwd_from.from_id, PeerUser):
         user = await u.User.get_by_tgid(TelegramID(fwd_from.from_id.user_id))
@@ -80,12 +109,11 @@ async def _add_forward_header(
                 )
 
         if not fwd_from_text:
-            try:
-                user = await source.client.get_entity(fwd_from.from_id)
-                if user:
-                    fwd_from_text, _ = pu.Puppet.get_displayname(user, False)
-                    fwd_from_html = f"<b>{escape(fwd_from_text)}</b>"
-            except (ValueError, RPCError):
+            user = await _get_fwd_entity(client, evt)
+            if user:
+                fwd_from_text, _ = pu.Puppet.get_displayname(user, False)
+                fwd_from_html = f"<b>{escape(fwd_from_text)}</b>"
+            else:
                 fwd_from_text = fwd_from_html = "unknown user"
     elif isinstance(fwd_from.from_id, (PeerChannel, PeerChat)):
         from_id = (
@@ -103,12 +131,11 @@ async def _add_forward_header(
             else:
                 fwd_from_html = f"channel <b>{escape(fwd_from_text)}</b>"
         else:
-            try:
-                channel = await source.client.get_entity(fwd_from.from_id)
-                if channel:
-                    fwd_from_text = f"channel {channel.title}"
-                    fwd_from_html = f"channel <b>{escape(channel.title)}</b>"
-            except (ValueError, RPCError):
+            channel = await _get_fwd_entity(client, evt)
+            if channel:
+                fwd_from_text = f"channel {channel.title}"
+                fwd_from_html = f"channel <b>{escape(channel.title)}</b>"
+            else:
                 fwd_from_text = fwd_from_html = "unknown channel"
     elif fwd_from.from_name:
         fwd_from_text = fwd_from.from_name
@@ -135,12 +162,14 @@ class ReuploadedCustomEmoji(MessageEntityCustomEmoji):
 
 
 async def _convert_custom_emoji(
-    source: au.AbstractUser, entities: list[TypeMessageEntity]
+    source: au.AbstractUser,
+    entities: list[TypeMessageEntity],
+    client: MautrixTelegramClient | None = None,
 ) -> None:
     emoji_ids = [
         entity.document_id for entity in entities if isinstance(entity, MessageEntityCustomEmoji)
     ]
-    custom_emojis = await transfer_custom_emojis_to_matrix(source, emoji_ids)
+    custom_emojis = await transfer_custom_emojis_to_matrix(source, emoji_ids, client=client)
     if len(custom_emojis) > 0:
         for i, entity in enumerate(entities):
             if isinstance(entity, MessageEntityCustomEmoji):
@@ -150,17 +179,20 @@ async def _convert_custom_emoji(
 async def telegram_to_matrix(
     evt: Message | SponsoredMessage,
     source: au.AbstractUser,
+    client: MautrixTelegramClient | None = None,
     override_text: str = None,
     override_entities: list[TypeMessageEntity] = None,
     require_html: bool = False,
 ) -> TextMessageEventContent:
+    if not client:
+        client = source.client
     content = TextMessageEventContent(
         msgtype=MessageType.TEXT,
         body=override_text or evt.message,
     )
     entities = override_entities or evt.entities
     if entities:
-        await _convert_custom_emoji(source, entities)
+        await _convert_custom_emoji(source, entities, client=client)
         content.format = Format.HTML
         html = await _telegram_entities_to_matrix_catch(add_surrogate(content.body), entities)
         content.formatted_body = del_surrogate(html)
@@ -169,7 +201,7 @@ async def telegram_to_matrix(
         content.ensure_has_html()
 
     if getattr(evt, "fwd_from", None):
-        await _add_forward_header(source, content, evt.fwd_from)
+        await _add_forward_header(client, content, evt)
 
     if isinstance(evt, Message) and evt.post and evt.post_author:
         content.ensure_has_html()
@@ -279,10 +311,14 @@ async def _telegram_entities_to_matrix(
         elif entity_type == MessageEntityCustomEmoji:
             html.append(entity_text)
         elif entity_type == ReuploadedCustomEmoji:
-            html.append(
-                f'<img data-mx-emoticon data-mau-animated-emoji src="{escape(entity.file.mxc)}" '
-                f'height="32" width="32" alt="{entity_text}" title="{entity_text}"/>'
-            )
+            if isinstance(entity.file, UnicodeCustomEmoji):
+                html.append(entity.file.emoji)
+            else:
+                html.append(
+                    f"<img data-mx-emoticon data-mau-animated-emoji"
+                    f' src="{escape(entity.file.mxc)}" height="32" width="32"'
+                    f' alt="{entity_text}" title="{entity_text}"/>'
+                )
         elif entity_type in (
             MessageEntityBotCommand,
             MessageEntityHashtag,

mautrix_telegram/matrix.py

@@ -16,15 +16,14 @@
 from __future__ import annotations
 
 from typing import TYPE_CHECKING
+import sys
 
 from mautrix.bridge import BaseMatrixHandler
-from mautrix.errors import MatrixError
 from mautrix.types import (
     Event,
     EventID,
     EventType,
     MemberStateEventContent,
-    MessageType,
     PresenceEvent,
     PresenceState,
     ReactionEvent,
@@ -36,7 +35,6 @@ from mautrix.types import (
     RoomTopicStateEventContent as TopicContent,
     SingleReceiptEventContent,
     StateEvent,
-    TextMessageEventContent,
     TypingEvent,
     UserID,
 )
@@ -63,6 +61,21 @@ class MatrixHandler(BaseMatrixHandler):
 
         self._previously_typing = {}
 
+    async def check_versions(self) -> None:
+        await super().check_versions()
+        if self.config["bridge.backfill.msc2716"] and not (
+            support := self.versions.supports("org.matrix.msc2716")
+        ):
+            self.log.fatal(
+                "Backfilling with MSC2716 is enabled in bridge config, but "
+                + (
+                    "batch sending is not enabled on homeserver"
+                    if support is False
+                    else "homeserver does not support batch sending"
+                )
+            )
+            sys.exit(18)
+
     async def handle_puppet_group_invite(
         self,
         room_id: RoomID,

mautrix_telegram/portal_util/message_convert.py

@@ -18,6 +18,7 @@ from __future__ import annotations
 from typing import Any, NamedTuple
 import base64
 import codecs
+import hashlib
 import html
 import mimetypes
 import unicodedata
@@ -63,6 +64,7 @@ from telethon.utils import decode_waveform
 
 from mautrix.appservice import IntentAPI
 from mautrix.types import (
+    EventID,
     EventType,
     Format,
     ImageInfo,
@@ -78,6 +80,7 @@ from mautrix.util.logging import TraceLogger
 from .. import abstract_user as au, formatter, matrix as m, portal as po, puppet as pu, util
 from ..config import Config
 from ..db import Message as DBMessage, TelegramFile as DBTelegramFile
+from ..tgclient import MautrixTelegramClient
 from ..types import TelegramID
 from ..util import sane_mimetypes
 
@@ -149,12 +152,16 @@ class TelegramMessageConverter:
         is_bot: bool,
         evt: Message,
         no_reply_fallback: bool = False,
+        deterministic_reply_id: bool = False,
+        client: MautrixTelegramClient | None = None,
     ) -> ConvertedMessage | None:
+        if not client:
+            client = source.client
         if hasattr(evt, "media") and isinstance(evt.media, self._allowed_media):
             convert_media = self._media_converters[type(evt.media)]
-            converted = await convert_media(source=source, intent=intent, evt=evt)
+            converted = await convert_media(source=source, intent=intent, evt=evt, client=client)
         elif evt.message:
-            converted = await self._convert_text(source, intent, is_bot, evt)
+            converted = await self._convert_text(source, intent, is_bot, evt, client)
         else:
             self.log.debug("Unhandled Telegram message %d", evt.id)
             return
@@ -176,7 +183,13 @@ class TelegramMessageConverter:
                 converted.caption.external_url = converted.content.external_url
                 if self.portal.get_config("caption_in_message"):
                     self._caption_to_message(converted)
-            await self._set_reply(source, evt, converted.content, no_fallback=no_reply_fallback)
+            await self._set_reply(
+                source,
+                evt,
+                converted.content,
+                no_fallback=no_reply_fallback,
+                deterministic_id=deterministic_reply_id,
+            )
         return converted
 
     @staticmethod
@@ -223,12 +236,19 @@ class TelegramMessageConverter:
             raise ValueError("Portal has invalid peer type")
         return base64.b64encode(play_id).decode("utf-8").rstrip("=")
 
+    def deterministic_event_id(self, space: TelegramID, msg_id: TelegramID) -> EventID:
+        hash_content = f"{self.portal.mxid}/telegram/{space}/{msg_id}"
+        hashed = hashlib.sha256(hash_content.encode("utf-8")).digest()
+        b64hash = base64.urlsafe_b64encode(hashed).decode("utf-8").rstrip("=")
+        return EventID(f"${b64hash}:telegram.org")
+
     async def _set_reply(
         self,
         source: au.AbstractUser,
         evt: Message,
         content: MessageEventContent,
         no_fallback: bool = False,
+        deterministic_id: bool = False,
     ) -> None:
         if not evt.reply_to:
             return
@@ -237,8 +257,11 @@ class TelegramMessageConverter:
             if isinstance(evt, Message) and isinstance(evt.peer_id, PeerChannel)
             else source.tgid
         )
-        msg = await DBMessage.get_one_by_tgid(TelegramID(evt.reply_to.reply_to_msg_id), space)
+        reply_to_id = TelegramID(evt.reply_to.reply_to_msg_id)
+        msg = await DBMessage.get_one_by_tgid(reply_to_id, space)
         if not msg or msg.mx_room != self.portal.mxid:
+            if deterministic_id:
+                content.set_reply(self.deterministic_event_id(space, reply_to_id))
             return
         elif not isinstance(content, TextMessageEventContent) or no_fallback:
             # Not a text message, just set the reply metadata and return
@@ -326,9 +349,14 @@ class TelegramMessageConverter:
         return beeper_link_preview
 
     async def _convert_text(
-        self, source: au.AbstractUser, intent: IntentAPI, is_bot: bool, evt: Message
+        self,
+        source: au.AbstractUser,
+        intent: IntentAPI,
+        is_bot: bool,
+        evt: Message,
+        client: MautrixTelegramClient,
     ) -> ConvertedMessage:
-        content = await formatter.telegram_to_matrix(evt, source)
+        content = await formatter.telegram_to_matrix(evt, source, client)
         if is_bot and self.portal.get_config("bot_messages_as_notices"):
             content.msgtype = MessageType.NOTICE
 
@@ -344,7 +372,11 @@ class TelegramMessageConverter:
         return ConvertedMessage(content=content)
 
     async def _convert_photo(
-        self, source: au.AbstractUser, intent: IntentAPI, evt: Message
+        self,
+        source: au.AbstractUser,
+        intent: IntentAPI,
+        evt: Message,
+        client: MautrixTelegramClient,
     ) -> ConvertedMessage | None:
         media: MessageMediaPhoto = evt.media
         if media.photo is None and media.ttl_seconds:
@@ -362,7 +394,7 @@ class TelegramMessageConverter:
                 )
             )
         file = await util.transfer_file_to_matrix(
-            source.client,
+            client,
             intent,
             loc,
             encrypt=self.portal.encrypted,
@@ -377,6 +409,8 @@ class TelegramMessageConverter:
             mimetype=file.mime_type,
             size=self._photo_size_key(largest_size),
         )
+        if media.spoiler:
+            info["fi.mau.telegram.spoiler"] = True
         ext = sane_mimetypes.guess_extension(file.mime_type)
         name = f"disappearing_image{ext}" if media.ttl_seconds else f"image{ext}"
         content = MediaMessageEventContent(
@@ -388,7 +422,9 @@ class TelegramMessageConverter:
             content.file = file.decryption_info
         else:
             content.url = file.mxc
-        caption_content = await formatter.telegram_to_matrix(evt, source) if evt.message else None
+        caption_content = (
+            await formatter.telegram_to_matrix(evt, source, client) if evt.message else None
+        )
         return ConvertedMessage(
             content=content,
             caption=caption_content,
@@ -396,7 +432,11 @@ class TelegramMessageConverter:
         )
 
     async def _convert_document(
-        self, source: au.AbstractUser, intent: IntentAPI, evt: Message
+        self,
+        source: au.AbstractUser,
+        intent: IntentAPI,
+        evt: Message,
+        client: MautrixTelegramClient,
     ) -> ConvertedMessage | None:
         document = evt.media.document
 
@@ -419,7 +459,7 @@ class TelegramMessageConverter:
         parallel_id = source.tgid if self.config["bridge.parallel_file_transfer"] else None
         tgs_convert = self.config["bridge.animated_sticker"]
         file = await util.transfer_file_to_matrix(
-            source.client,
+            client,
             intent,
             document,
             thumb_loc,
@@ -460,8 +500,10 @@ class TelegramMessageConverter:
             info["fi.mau.autoplay"] = True
             info["fi.mau.hide_controls"] = True
             info["fi.mau.no_audio"] = True
+        if evt.media.spoiler:
+            info["fi.mau.telegram.spoiler"] = True
         if not name:
-            ext = sane_mimetypes.guess_extension(file.mime_type)
+            ext = sane_mimetypes.guess_extension(file.mime_type) or ""
             name = "unnamed_file" + ext
 
         content = MediaMessageEventContent(
@@ -486,7 +528,9 @@ class TelegramMessageConverter:
         else:
             content.url = file.mxc
 
-        caption_content = await formatter.telegram_to_matrix(evt, source) if evt.message else None
+        caption_content = (
+            await formatter.telegram_to_matrix(evt, source, client) if evt.message else None
+        )
 
         return ConvertedMessage(
             type=event_type,
@@ -527,13 +571,17 @@ class TelegramMessageConverter:
         return ConvertedMessage(content=content)
 
     @staticmethod
-    async def _convert_unsupported(source: au.AbstractUser, evt: Message, **_) -> ConvertedMessage:
+    async def _convert_unsupported(
+        source: au.AbstractUser, evt: Message, client: MautrixTelegramClient, **_
+    ) -> ConvertedMessage:
         override_text = (
             "This message is not supported on your version of Mautrix-Telegram. "
             "Please check https://github.com/mautrix/telegram or ask your "
             "bridge administrator about possible updates."
         )
-        content = await formatter.telegram_to_matrix(evt, source, override_text=override_text)
+        content = await formatter.telegram_to_matrix(
+            evt, source, client, override_text=override_text
+        )
         content.msgtype = MessageType.NOTICE
         content["fi.mau.telegram.unsupported"] = True
         return ConvertedMessage(content=content)
@@ -589,7 +637,9 @@ class TelegramMessageConverter:
         content["fi.mau.telegram.dice"] = {"emoticon": roll.emoticon, "value": roll.value}
         return ConvertedMessage(content=content)
 
-    async def _convert_game(self, source: au.AbstractUser, evt: Message, **_) -> ConvertedMessage:
+    async def _convert_game(
+        self, source: au.AbstractUser, evt: Message, client: MautrixTelegramClient, **_
+    ) -> ConvertedMessage:
         game: Game = evt.media.game
         play_id = self._encode_msgid(source, evt)
         command = f"{self.command_prefix} play {play_id}"
@@ -599,7 +649,7 @@ class TelegramMessageConverter:
         ]
 
         content = await formatter.telegram_to_matrix(
-            evt, source, override_text=override_text, override_entities=override_entities
+            evt, source, client, override_text=override_text, override_entities=override_entities
         )
         content.msgtype = MessageType.NOTICE
         content["fi.mau.telegram.game"] = play_id
@@ -607,7 +657,9 @@ class TelegramMessageConverter:
         return ConvertedMessage(content=content)
 
     @staticmethod
-    async def _convert_contact(source: au.AbstractUser, evt: Message, **_) -> ConvertedMessage:
+    async def _convert_contact(
+        source: au.AbstractUser, evt: Message, client: MautrixTelegramClient, **_
+    ) -> ConvertedMessage:
         contact: MessageMediaContact = evt.media
         name = " ".join(x for x in [contact.first_name, contact.last_name] if x)
         formatted_phone = f"+{contact.phone_number}"
@@ -633,8 +685,8 @@ class TelegramMessageConverter:
         puppet = await pu.Puppet.get_by_tgid(TelegramID(contact.user_id))
         if not puppet.displayname:
             try:
-                entity = await source.client.get_entity(PeerUser(contact.user_id))
-                await puppet.update_info(source, entity)
+                entity = await client.get_entity(PeerUser(contact.user_id))
+                await puppet.update_info(source, entity, client_override=client)
             except Exception as e:
                 source.log.warning(f"Failed to sync puppet info of received contact: {e}")
         else:

mautrix_telegram/portal_util/power_levels.py

@@ -34,8 +34,12 @@ from ..types import TelegramID
 
 
 def get_base_power_levels(
-    portal: po.Portal, levels: PowerLevelContent = None, entity: TypeChat = None
+    portal: po.Portal,
+    levels: PowerLevelContent = None,
+    entity: TypeChat | None = None,
+    dbr: ChatBannedRights | None = None,
 ) -> PowerLevelContent:
+    is_initial = not levels
     levels = levels or PowerLevelContent()
     if portal.peer_type == "user":
         overrides = portal.config["bridge.initial_power_level_overrides.user"]
@@ -51,7 +55,7 @@ def get_base_power_levels(
         levels.events_default = overrides.get("events_default", 0)
     else:
         overrides = portal.config["bridge.initial_power_level_overrides.group"]
-        dbr = entity.default_banned_rights
+        dbr = dbr or entity.default_banned_rights
         if not dbr:
             portal.log.debug(f"default_banned_rights is None in {entity}")
             dbr = ChatBannedRights(
@@ -80,16 +84,14 @@ def get_base_power_levels(
         levels.events_default = overrides.get(
             "events_default",
             50
-            if portal.peer_type == "channel" and not entity.megagroup or dbr.send_messages
+            if portal.peer_type == "channel" and not portal.megagroup or dbr.send_messages
             else 0,
         )
     for evt_type, value in overrides.get("events", {}).items():
         levels.events[EventType.find(evt_type)] = value
     userlevel_overrides = overrides.get("users", {})
-    bot_level = levels.get_user_level(portal.main_intent.mxid)
-    for user, user_level in levels.users.items():
-        if user_level < bot_level:
-            levels.users[user] = userlevel_overrides.get(user, 0)
+    if is_initial:
+        levels.users.update(userlevel_overrides)
     if portal.main_intent.mxid not in levels.users:
         levels.users[portal.main_intent.mxid] = 100
     return levels

mautrix_telegram/portal_util/sponsored_message.py

@@ -20,6 +20,7 @@ import html
 
 from telethon.tl.functions.channels import GetSponsoredMessagesRequest
 from telethon.tl.types import Channel, InputChannel, PeerChannel, PeerUser, SponsoredMessage, User
+from telethon.tl.types.messages import SponsoredMessages, SponsoredMessagesEmpty
 
 from mautrix.types import MessageType, TextMessageEventContent
 
@@ -32,8 +33,9 @@ async def get_sponsored_message(
     entity: InputChannel,
 ) -> tuple[SponsoredMessage | None, int | None, Channel | User | None]:
     resp = await user.client(GetSponsoredMessagesRequest(entity))
-    if len(resp.messages) == 0:
+    if isinstance(resp, SponsoredMessagesEmpty):
         return None, None, None
+    assert isinstance(resp, SponsoredMessages)
     msg = resp.messages[0]
     if isinstance(msg.from_id, PeerUser):
         entities = resp.users

mautrix_telegram/puppet.py

@@ -19,6 +19,7 @@ from typing import TYPE_CHECKING, AsyncGenerator, AsyncIterable, Awaitable, cast
 from difflib import SequenceMatcher
 import unicodedata
 
+from telethon import utils
 from telethon.tl.types import (
     Channel,
     ChatPhoto,
@@ -29,8 +30,6 @@ from telethon.tl.types import (
     PeerChat,
     PeerUser,
     TypeChatPhoto,
-    TypeInputPeer,
-    TypeInputUser,
     TypePeer,
     TypeUserProfilePhoto,
     UpdateUserName,
@@ -48,6 +47,7 @@ from mautrix.util.simple_template import SimpleTemplate
 from . import abstract_user as au, portal as p, util
 from .config import Config
 from .db import Puppet as DBPuppet
+from .tgclient import MautrixTelegramClient
 from .types import TelegramID
 
 if TYPE_CHECKING:
@@ -147,9 +147,6 @@ class Puppet(DBPuppet, BasePuppet):
     def plain_displayname(self) -> str:
         return self.displayname_template.parse(self.displayname) or self.displayname
 
-    def get_input_entity(self, user: au.AbstractUser) -> Awaitable[TypeInputPeer | TypeInputUser]:
-        return user.client.get_input_entity(self.peer)
-
     def intent_for(self, portal: p.Portal) -> IntentAPI:
         if portal.tgid == self.tgid:
             return self.default_mxid_intent
@@ -255,7 +252,12 @@ class Puppet(DBPuppet, BasePuppet):
         except Exception:
             source.log.exception(f"Failed to update info of {self.tgid}")
 
-    async def update_info(self, source: au.AbstractUser, info: User | Channel) -> None:
+    async def update_info(
+        self,
+        source: au.AbstractUser,
+        info: User | Channel,
+        client_override: MautrixTelegramClient | None = None,
+    ) -> None:
         is_bot = False if isinstance(info, Channel) else info.bot
         is_premium = False if isinstance(info, Channel) else info.premium
         is_channel = isinstance(info, Channel)
@@ -277,8 +279,16 @@ class Puppet(DBPuppet, BasePuppet):
 
         if not self.disable_updates:
             try:
-                changed = await self.update_displayname(source, info) or changed
-                changed = await self.update_avatar(source, info.photo) or changed
+                changed = (
+                    await self.update_displayname(source, info, client_override=client_override)
+                    or changed
+                )
+                changed = (
+                    await self.update_avatar(
+                        source, info.photo, entity=info, client_override=client_override
+                    )
+                    or changed
+                )
             except Exception:
                 self.log.exception(f"Failed to update info from source {source.tgid}")
 
@@ -293,7 +303,10 @@ class Puppet(DBPuppet, BasePuppet):
             await portal.update_info_from_puppet(self)
 
     async def update_displayname(
-        self, source: au.AbstractUser, info: User | Channel | UpdateUserName
+        self,
+        source: au.AbstractUser,
+        info: User | Channel | UpdateUserName,
+        client_override: MautrixTelegramClient | None = None,
     ) -> bool:
         if self.disable_updates:
             return False
@@ -320,14 +333,16 @@ class Puppet(DBPuppet, BasePuppet):
             return False
 
         if isinstance(info, UpdateUserName):
-            info = await source.client.get_entity(self.peer)
-        if isinstance(info, Channel) or not info.contact:
-            self.displayname_contact = False
-        elif not self.displayname_contact:
-            if not self.displayname:
-                self.displayname_contact = True
-            else:
-                return False
+            info = await (client_override or source.client).get_entity(self.peer)
+        is_contact_name = not isinstance(info, Channel) and info.contact
+        # Reject name change if the contact status is moving in an unwanted direction,
+        # and we already have a name for the ghost.
+        if (
+            is_contact_name != self.displayname_contact
+            and is_contact_name != self.config["bridge.allow_contact_info"]
+            and self.displayname
+        ):
+            return False
 
         displayname, quality = self.get_displayname(info)
         needs_reset = displayname != self.displayname or not self.name_set
@@ -335,12 +350,14 @@ class Puppet(DBPuppet, BasePuppet):
         if needs_reset and is_high_quality:
             allow_because = f"{allow_because} and quality {quality} >= {self.displayname_quality}"
             self.log.debug(
-                f"Updating displayname of {self.id} (src: {source.tgid}, allowed "
-                f"because {allow_because}) from {self.displayname} to {displayname}"
+                f"Updating displayname of {self.id} (src: {source.tgid}, "
+                f"contact: {is_contact_name}, allowed because {allow_because}) "
+                f"from {self.displayname} to {displayname}"
             )
             self.log.trace("Displayname source data: %s", info)
             self.displayname = displayname
             self.displayname_source = source.tgid
+            self.displayname_contact = is_contact_name
             self.displayname_quality = quality
             try:
                 await self.default_mxid_intent.set_displayname(
@@ -357,10 +374,24 @@ class Puppet(DBPuppet, BasePuppet):
         return False
 
     async def update_avatar(
-        self, source: au.AbstractUser, photo: TypeUserProfilePhoto | TypeChatPhoto
+        self,
+        source: au.AbstractUser,
+        photo: TypeUserProfilePhoto | TypeChatPhoto,
+        entity: User | None = None,
+        client_override: MautrixTelegramClient | None = None,
     ) -> bool:
         if self.disable_updates:
             return False
+        if (
+            isinstance(photo, UserProfilePhoto)
+            and photo.personal
+            and not self.config["bridge.allow_contact_info"]
+        ):
+            self.log.trace(
+                "Dropping user avatar as it's personal "
+                "and contact info is disabled in bridge config"
+            )
+            return False
 
         if photo is None or isinstance(photo, (UserProfilePhotoEmpty, ChatPhotoEmpty)):
             photo_id = ""
@@ -376,11 +407,22 @@ class Puppet(DBPuppet, BasePuppet):
                 self.photo_id = ""
                 self.avatar_url = None
             elif self.photo_id != photo_id or not self.avatar_url:
+                client = client_override or source.client
+                try:
+                    peer = await client.get_input_entity(entity or self.peer)
+                except ValueError:
+                    if entity:
+                        peer = utils.get_input_peer(entity, check_hash=False)
+                    else:
+                        self.log.warning(f"Couldn't get input entity to update avatar")
+                        return False
                 file = await util.transfer_file_to_matrix(
-                    client=source.client,
+                    client=client,
                     intent=self.default_mxid_intent,
                     location=InputPeerPhotoFileLocation(
-                        peer=await self.get_input_entity(source), photo_id=photo.photo_id, big=True
+                        peer=peer,
+                        photo_id=photo.photo_id,
+                        big=True,
                     ),
                     async_upload=self.config["homeserver.async_media"],
                 )
@@ -399,7 +441,7 @@ class Puppet(DBPuppet, BasePuppet):
 
     async def default_puppet_should_leave_room(self, room_id: RoomID) -> bool:
         portal: p.Portal = await p.Portal.get_by_mxid(room_id)
-        return portal and not portal.backfill_lock.locked and portal.peer_type != "user"
+        return portal and portal.peer_type != "user"
 
     # endregion
     # region Getters

mautrix_telegram/scripts/unicodemojipack/__main__.py

@@ -0,0 +1,397 @@
+# mautrix-telegram - A Matrix-Telegram puppeting bridge
+# Copyright (C) 2022 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from typing import Any, Literal, TypedDict
+from pathlib import Path
+import argparse
+import asyncio
+import io
+import json
+import logging
+import math
+import mimetypes
+import pickle
+import random
+import string
+
+from lottie.exporters import export_tgs
+from lottie.exporters.cairo import export_png
+from lottie.exporters.tgs_validator import Severity, TgsValidator
+from lottie.importers.svg import import_svg
+from lottie.objects import Animation
+from lottie.utils.stripper import float_strip
+from PIL import Image
+from telethon import TelegramClient
+from telethon.custom import Conversation, Message
+from telethon.tl.functions.messages import GetStickerSetRequest
+from telethon.tl.types import (
+    Document,
+    DocumentAttributeCustomEmoji,
+    DocumentAttributeFilename,
+    DocumentAttributeImageSize,
+    InputMediaUploadedDocument,
+    InputStickerSetShortName,
+)
+import aiohttp
+
+mimetypes.add_type("image/webp", ".webp")
+
+parser = argparse.ArgumentParser(description="mautrix-telegram unicode emoji packer")
+parser.add_argument(
+    "-i", "--api-id", type=int, required=True, metavar="<api id>", help="Telegram API ID"
+)
+parser.add_argument(
+    "-a", "--api-hash", type=str, required=True, metavar="<api hash>", help="Telegram API hash"
+)
+parser.add_argument(
+    "-s",
+    "--session",
+    type=str,
+    default="unicodemojipacker.session",
+    metavar="<file name>",
+    help="Telethon session name",
+)
+parser.add_argument(
+    "-o",
+    "--output",
+    type=str,
+    default="mautrix_telegram/unicodemojipack.json",
+    metavar="<file name>",
+    help="Path to save created emoji pack document IDs",
+)
+parser.add_argument(
+    "-f",
+    "--font-directory",
+    type=Path,
+    required=True,
+    metavar="<directory path>",
+    help="Path to the Noto color emoji files",
+)
+parser.add_argument(
+    "-m",
+    "--media-directory",
+    type=Path,
+    required=True,
+    metavar="<directory path>",
+    help="Path to save converted tgs and webp emoji files",
+)
+args = parser.parse_args()
+font_dir: Path = args.font_directory
+media_dir: Path = args.media_directory
+
+EMOJI_DATA_URL = "https://raw.githubusercontent.com/iamcal/emoji-data/master/emoji.json"
+
+
+def unified_to_unicode(unified: str) -> str:
+    return (
+        "".join(rf"\U{chunk:0>8}" for chunk in unified.split("-"))
+        .encode("ascii")
+        .decode("unicode_escape")
+    )
+
+
+def tag_to_str(unified: str) -> str:
+    return "".join(chr(int(x.removeprefix("E00"), 16)) for x in unified.split("-"))
+
+
+EmojiType = Literal["webp", "tgs"]
+PackType = Literal["Animated emoji", "Static emoji"]
+
+
+class Emoji(TypedDict):
+    hex: str
+    emoji: str
+    type: EmojiType
+    filename: str
+
+
+class EmojiData(TypedDict):
+    tgs: list[Emoji]
+    webp: list[Emoji]
+
+
+def parse_emoji_data(tone: dict[str, Any], emoji: dict[str, Any]) -> Emoji:
+    hex = (tone["non_qualified"] or tone["unified"]).replace("-FE0F", "")
+    filename_hex = hex.replace("-", "_").lower()
+    filename = f"svg/emoji_u{filename_hex}.svg"
+    if emoji["category"] == "Flags" and emoji["subcategory"] in (
+        "country-flag",
+        "subdivision-flag",
+    ):
+        filename = f"third_party/region-flags/waved-svg/emoji_u{filename_hex}.svg"
+
+    with (font_dir / filename).open() as f:
+        lot: Animation = import_svg(f)
+    float_strip(lot)
+    lot.tgs_sanitize()
+
+    output = io.BytesIO()
+    export_tgs(lot, output)
+
+    validator = TgsValidator()
+    validator(lot)
+    validator.check_size(len(output.getvalue()))
+    errors = [err for err in validator.errors if err.severity != Severity.Note]
+    if errors or ("region-flags" in filename and len(output.getvalue()) > 32768):
+        lot.scale(100, 100)
+
+        png_out = io.BytesIO()
+        export_png(lot, png_out)
+        img = Image.open(png_out)
+        output = io.BytesIO()
+        output.name = "image.webp"
+        img.save(output, "webp")
+
+        media_type: EmojiType = "webp"
+    else:
+        media_type: EmojiType = "tgs"
+    path = media_dir / f"{filename_hex}.{media_type}"
+    with path.open("wb") as f:
+        f.write(output.getvalue())
+    print(
+        "Converted", filename, "->", path.name, "//" if errors else "", "\n".join(map(str, errors))
+    )
+
+    return {
+        "hex": hex,
+        "emoji": unified_to_unicode(tone["unified"]),
+        "type": media_type,
+        "filename": path.name,
+    }
+
+
+async def load_emoji_data() -> EmojiData:
+    cache_path = media_dir / "conversion-cache.json"
+    try:
+        with cache_path.open() as f:
+            return json.load(f)
+    except FileNotFoundError:
+        pass
+    async with aiohttp.ClientSession() as sess, sess.get(EMOJI_DATA_URL) as resp:
+        raw_emoji_data = sorted(
+            await resp.json(content_type=None),
+            key=lambda dat: dat["sort_order"],
+        )
+    tgs_emoji = []
+    webp_emoji = []
+    for emoji in raw_emoji_data:
+        for tone in (emoji, *emoji.get("skin_variations", {}).values()):
+            parsed_emoji = parse_emoji_data(tone, emoji)
+            if parsed_emoji["type"] == "tgs":
+                tgs_emoji.append(parsed_emoji)
+            else:
+                webp_emoji.append(parsed_emoji)
+    full_data = {"tgs": tgs_emoji, "webp": webp_emoji}
+    with cache_path.open("w") as f:
+        json.dump(full_data, f, ensure_ascii=False)
+    return full_data
+
+
+async def create_pack(conv: Conversation, name: str, pack_type: str) -> None:
+    await conv.send_message("/newemojipack")
+    resp: Message = await conv.get_response()
+    assert "A new set of custom emoji" in resp.raw_text
+    assert "Please choose the type" in resp.raw_text
+    await conv.send_message(pack_type)
+    resp = await conv.get_response()
+    if pack_type == "Animated emoji":
+        assert "When ready to upload, tell me the name of your set." in resp.raw_text
+    else:
+        assert "Now choose a name for your set." in resp.raw_text
+    await conv.send_message(name)
+    resp = await conv.get_response()
+    if pack_type == "Animated emoji":
+        assert "Now send me the first animated emoji" in resp.raw_text
+    else:
+        assert "Now send me the custom emoji" in resp.raw_text
+
+
+async def publish_pack(conv: Conversation, shortname: str) -> None:
+    await conv.send_message("/publish")
+
+    resp: Message = await conv.get_response()
+    assert "You can send me a custom emoji from your emoji set" in resp.raw_text
+    await conv.send_message("/skip")
+
+    resp = await conv.get_response()
+    assert "Please provide a short name for your emoji set" in resp.raw_text
+    await conv.send_message(shortname)
+
+    resp = await conv.get_response()
+    assert "I've just published your emoji set" in resp.raw_text
+
+
+async def send_emoji(
+    conv: Conversation, file: bytes | Path | InputMediaUploadedDocument, emoji: str
+) -> None:
+    await conv.send_file(file)
+    resp: Message = await conv.get_response()
+    assert "Send me a replacement emoji that corresponds to your custom emoji" in resp.raw_text
+    await conv.send_message(emoji)
+    resp = await conv.get_response()
+    if "Sorry, too many attempts" in resp.raw_text:
+        print(resp.raw_text)
+        input("Press enter to continue")
+        await conv.send_message(emoji)
+        resp = await conv.get_response()
+    while "Please send an emoji that best describes your custom emoji." in resp.raw_text:
+        emoji = input(f"{emoji} was rejected, provide replacement: ")
+        await conv.send_message(emoji)
+        resp = await conv.get_response()
+    assert "Congratulations" in resp.raw_text
+
+
+class CachedPack(TypedDict):
+    name: str
+    short_name: str
+    part: int
+    type: PackType
+    published: bool
+    collected: bool
+    emojis: list[Emoji]
+
+
+class CachedData(TypedDict):
+    packs: list[CachedPack]
+
+
+def _split_packs_int(
+    emoji_list: list[Emoji], pack_type: PackType, current_part: int, total_parts: int
+) -> tuple[list[CachedPack], int]:
+    packs = []
+    current_pack: CachedPack | None = None
+    for i, emoji in enumerate(emoji_list):
+        if i % 200 == 0:
+            current_part += 1
+            random_id = "".join(random.choices(string.ascii_lowercase + string.digits, k=8))
+            short_name = f"mxtg_unicodemoji_{random_id}"
+            name = f"mautrix-telegram unicodemoji ({current_part}/{total_parts})"
+            current_pack = {
+                "type": pack_type,
+                "short_name": short_name,
+                "part": current_part,
+                "name": name,
+                "published": False,
+                "collected": False,
+                "emojis": [],
+            }
+            packs.append(current_pack)
+        current_pack["emojis"].append(emoji)
+    return packs, current_part
+
+
+def split_packs(emoji_data: EmojiData) -> list[CachedPack]:
+    total_parts = math.ceil(len(emoji_data["tgs"]) / 200) + math.ceil(
+        len(emoji_data["webp"]) / 200
+    )
+    current_part = 0
+    animated_packs, current_part = _split_packs_int(
+        emoji_data["tgs"], "Animated emoji", current_part, total_parts
+    )
+    static_packs, current_part = _split_packs_int(
+        emoji_data["webp"], "Static emoji", current_part, total_parts
+    )
+    return animated_packs + static_packs
+
+
+async def create_and_fill_pack(
+    client: TelegramClient, conv: Conversation, pack: CachedPack
+) -> None:
+    if pack["short_name"] == "mxtg_unicodemoji_xvzs6743":
+        print("Continuing pack", pack["name"])
+    else:
+        print("Creating pack", pack["name"])
+        await create_pack(conv, pack["name"], pack["type"])
+    total = len(pack["emojis"])
+    for i, emoji in enumerate(pack["emojis"]):
+        if pack["short_name"] == "mxtg_unicodemoji_xvzs6743" and i < 87:
+            continue
+        print(f"Adding emoji {i+1}/{total}", emoji["hex"], emoji["emoji"])
+        emoji_file = media_dir / emoji["filename"]
+        if emoji["type"] == "webp":
+            attrs = [
+                DocumentAttributeImageSize(w=100, h=100),
+                DocumentAttributeFilename(file_name="image.webp"),
+            ]
+            with emoji_file.open("rb") as f:
+                file_handle = await client.upload_file(f, file_name="emoji.webp")
+            emoji_file = InputMediaUploadedDocument(
+                file_handle, mime_type="image/webp", attributes=attrs
+            )
+        await send_emoji(conv, emoji_file, emoji["emoji"])
+        await asyncio.sleep(2)
+    print("Publishing pack", pack["short_name"])
+    await publish_pack(conv, pack["short_name"])
+
+
+async def main():
+    logging.basicConfig(level=logging.INFO)
+
+    emoji_data = await load_emoji_data()
+
+    split_cache = media_dir / "split-cache.json"
+    try:
+        with split_cache.open() as f:
+            packs: list[CachedPack] = json.load(f)
+    except FileNotFoundError:
+        packs = split_packs(emoji_data)
+        with split_cache.open("w") as f:
+            json.dump(packs, f)
+
+    doc_id_file = Path(args.output)
+    try:
+        with doc_id_file.open() as f:
+            doc_ids = json.load(f)
+    except FileNotFoundError:
+        doc_ids = {}
+
+    client = TelegramClient(args.session, args.api_id, args.api_hash, flood_sleep_threshold=3600)
+    await client.start()
+    async with client.conversation("Stickers", max_messages=20000) as conv:
+        for pack in packs:
+            if not pack["published"]:
+                await create_and_fill_pack(client, conv, pack)
+                pack["published"] = True
+                with split_cache.open("w") as f:
+                    json.dump(packs, f, ensure_ascii=False)
+            if not pack["collected"] or True:
+                print("Collecting document IDs from pack", pack["short_name"])
+                stickers = await client(
+                    GetStickerSetRequest(InputStickerSetShortName(pack["short_name"]), 0)
+                )
+                doc: Document
+                for i, doc in enumerate(stickers.documents):
+                    attr = next(
+                        attr
+                        for attr in doc.attributes
+                        if isinstance(attr, DocumentAttributeCustomEmoji)
+                    )
+                    base_emoji = attr.alt.replace("\ufe0f", "")
+                    emoji = pack["emojis"][i]["emoji"].replace("\ufe0f", "")
+                    doc_ids[emoji] = doc.id
+                    print(f"Mapped {emoji} (fallback: {base_emoji}) -> {doc_ids[emoji]}")
+                pack["collected"] = True
+                with split_cache.open("w") as f:
+                    json.dump(packs, f, ensure_ascii=False)
+                with doc_id_file.open("w") as f:
+                    json.dump(doc_ids, f, ensure_ascii=False)
+                print("Pack completed")
+                await asyncio.sleep(5)
+    with open(args.output.replace(".json", ".pickle"), "wb") as f:
+        pickle.dump(doc_ids, f)
+    print("Wrote pickle")
+
+
+asyncio.run(main())

mautrix_telegram/user.py

@@ -15,21 +15,30 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, AsyncGenerator, AsyncIterable, Awaitable, NamedTuple, cast
-from datetime import datetime, timezone
+from typing import TYPE_CHECKING, Any, AsyncGenerator, AsyncIterable, Awaitable, NamedTuple, cast
+from datetime import datetime
 import asyncio
+import time
 
-from telethon.errors import AuthKeyDuplicatedError, RPCError, UnauthorizedError
+from telethon.errors import (
+    AuthKeyDuplicatedError,
+    AuthKeyError,
+    RPCError,
+    TakeoutInitDelayError,
+    UnauthorizedError,
+)
 from telethon.tl.custom import Dialog
 from telethon.tl.functions.account import UpdateStatusRequest
 from telethon.tl.functions.contacts import GetContactsRequest, SearchRequest
+from telethon.tl.functions.help import GetAppConfigRequest
+from telethon.tl.functions.messages import GetAvailableReactionsRequest
 from telethon.tl.functions.updates import GetStateRequest
 from telethon.tl.functions.users import GetUsersRequest
 from telethon.tl.types import (
-    Channel,
     Chat,
     ChatForbidden,
     InputUserSelf,
+    Message,
     NotifyPeer,
     PeerUser,
     TypeUpdate,
@@ -43,18 +52,21 @@ from telethon.tl.types import (
     User as TLUser,
 )
 from telethon.tl.types.contacts import ContactsNotModified
+from telethon.tl.types.messages import AvailableReactions
 
 from mautrix.appservice import DOUBLE_PUPPET_SOURCE_KEY
 from mautrix.bridge import BaseUser, async_getter_lock
 from mautrix.client import Client
 from mautrix.errors import MatrixRequestError, MNotFound
 from mautrix.types import PushActionType, PushRuleKind, PushRuleScope, RoomID, RoomTagInfo, UserID
+from mautrix.util import background_task
 from mautrix.util.bridge_state import BridgeState, BridgeStateEvent
 from mautrix.util.opt_prometheus import Gauge
 
-from . import portal as po, puppet as pu
+from . import portal as po, puppet as pu, util
 from .abstract_user import AbstractUser
-from .db import Message as DBMessage, PgSession, User as DBUser
+from .db import Backfill, BackfillType, Message as DBMessage, PgSession, User as DBUser
+from .tgclient import MautrixTelegramClient
 from .types import TelegramID
 
 if TYPE_CHECKING:
@@ -83,7 +95,17 @@ class User(DBUser, AbstractUser, BaseUser):
 
     _ensure_started_lock: asyncio.Lock
     _track_connection_task: asyncio.Task | None
+    _backfill_task: asyncio.Task | None
+    wakeup_backfill_task: asyncio.Event
     _is_backfilling: bool
+    takeout_retry_immediate: asyncio.Event
+    takeout_requested: bool
+
+    _available_emoji_reactions: set[str] | None
+    _available_emoji_reactions_hash: int | None
+    _available_emoji_reactions_fetched: float
+    _available_emoji_reactions_lock: asyncio.Lock
+    _app_config: dict[str, Any] | None
 
     def __init__(
         self,
@@ -111,6 +133,17 @@ class User(DBUser, AbstractUser, BaseUser):
         self._is_backfilling = False
         self._portals_cache = None
 
+        self._backfill_task = None
+        self.wakeup_backfill_task = asyncio.Event()
+        self.takeout_retry_immediate = asyncio.Event()
+        self.takeout_requested = False
+
+        self._available_emoji_reactions = None
+        self._available_emoji_reactions_hash = None
+        self._available_emoji_reactions_fetched = 0
+        self._available_emoji_reactions_lock = asyncio.Lock()
+        self._app_config = None
+
         (
             self.relaybot_whitelisted,
             self.whitelisted,
@@ -176,17 +209,30 @@ class User(DBUser, AbstractUser, BaseUser):
         async with self._ensure_started_lock:
             return cast(User, await super().ensure_started(even_if_no_session))
 
+    async def on_signed_out(self, err: UnauthorizedError | AuthKeyError) -> None:
+        error_code = "tg-auth-error"
+        if isinstance(err, AuthKeyDuplicatedError):
+            error_code = "tg-auth-key-duplicated"
+            message = None
+        else:
+            message = str(err)
+        self.log.warning(f"User got signed out with {err}, deleting data...")
+        try:
+            await self.log_out(
+                state=BridgeStateEvent.BAD_CREDENTIALS,
+                error=error_code,
+                message=message,
+                delete=False,
+            )
+        except Exception:
+            self.log.exception("Error handling external logout")
+
     async def start(self, delete_unless_authenticated: bool = False) -> User:
         try:
             await super().start()
-        except AuthKeyDuplicatedError:
+        except AuthKeyDuplicatedError as e:
             self.log.warning("Got AuthKeyDuplicatedError in start()")
-            await self.push_bridge_state(
-                BridgeStateEvent.BAD_CREDENTIALS, error="tg-auth-key-duplicated"
-            )
-            await self.client.disconnect()
-            await self.client.session.delete()
-            self.client = None
+            await self.on_signed_out(e)
             if not delete_unless_authenticated:
                 # The caller wants the client to be connected, so restart the connection.
                 await super().start()
@@ -206,12 +252,7 @@ class User(DBUser, AbstractUser, BaseUser):
             if delete_unless_authenticated or self.tgid:
                 self.log.error(f"Authorization error in start(): {type(e)}: {e}")
             if self.tgid:
-                await self.push_bridge_state(
-                    BridgeStateEvent.BAD_CREDENTIALS,
-                    error="tg-auth-error",
-                    message=str(e),
-                    ttl=3600,
-                )
+                await self.on_signed_out(e)
         except RPCError as e:
             self.log.error(f"Unknown RPC error in start(): {type(e)}: {e}")
             if self.tgid:
@@ -219,10 +260,10 @@ class User(DBUser, AbstractUser, BaseUser):
         else:
             # Authenticated, run post login
             self.log.debug(f"Ensuring post_login() for {self.name}")
-            asyncio.create_task(self.post_login())
+            background_task.create(self.post_login())
             return self
         # Not authenticated, delete data if necessary
-        if delete_unless_authenticated:
+        if delete_unless_authenticated and self.client is not None:
             self.log.debug(f"Unauthenticated user {self.name} start()ed, deleting session...")
             await self.client.disconnect()
             await self.client.session.delete()
@@ -234,6 +275,14 @@ class User(DBUser, AbstractUser, BaseUser):
             self.client and self.client._sender and self.client._sender._transport_connected()
         )
 
+    @property
+    def _bridge_state_info(self) -> dict[str, Any]:
+        if self.takeout_requested:
+            return {
+                "takeout_requested": True,
+            }
+        return {}
+
     async def _track_connection(self) -> None:
         self.log.debug("Starting loop to track connection state")
         while True:
@@ -245,17 +294,18 @@ class User(DBUser, AbstractUser, BaseUser):
                     BridgeStateEvent.BACKFILLING
                     if self._is_backfilling
                     else BridgeStateEvent.CONNECTED,
-                    ttl=3600,
+                    info=self._bridge_state_info,
                 )
             else:
                 await self.push_bridge_state(
-                    BridgeStateEvent.TRANSIENT_DISCONNECT, ttl=240, error="tg-not-connected"
+                    BridgeStateEvent.TRANSIENT_DISCONNECT, error="tg-not-connected"
                 )
 
     async def fill_bridge_state(self, state: BridgeState) -> None:
         await super().fill_bridge_state(state)
-        state.remote_id = str(self.tgid)
-        state.remote_name = self.human_tg_id
+        if self.tgid:
+            state.remote_id = str(self.tgid)
+            state.remote_name = self.human_tg_id
 
     async def get_bridge_states(self) -> list[BridgeState]:
         if not self.tgid:
@@ -270,7 +320,7 @@ class User(DBUser, AbstractUser, BaseUser):
         else:
             state_event = BridgeStateEvent.UNKNOWN_ERROR
             ttl = 240
-        return [BridgeState(state_event=state_event, ttl=ttl)]
+        return [BridgeState(state_event=state_event, ttl=ttl, info=self._bridge_state_info)]
 
     async def get_puppet(self) -> pu.Puppet | None:
         if not self.tgid:
@@ -288,6 +338,9 @@ class User(DBUser, AbstractUser, BaseUser):
         if self._track_connection_task:
             self._track_connection_task.cancel()
             self._track_connection_task = None
+        if self._backfill_task:
+            self._backfill_task.cancel()
+            self._backfill_task = None
         await super().stop()
         self._track_metric(METRIC_CONNECTED, False)
 
@@ -304,6 +357,8 @@ class User(DBUser, AbstractUser, BaseUser):
             return
 
         self._track_metric(METRIC_LOGGED_IN, True)
+        if not self._backfill_task or self._backfill_task.done():
+            self._backfill_task = asyncio.create_task(self._try_handle_backfill_requests_loop())
 
         try:
             puppet = await pu.Puppet.get_by_tgid(self.tgid)
@@ -313,7 +368,7 @@ class User(DBUser, AbstractUser, BaseUser):
         except Exception:
             self.log.exception("Failed to automatically enable custom puppet")
 
-        if not self.is_bot and self.config["bridge.startup_sync"]:
+        if not self.is_bot and (self.config["bridge.startup_sync"] or first_login):
             try:
                 self._is_backfilling = True
                 await self.sync_dialogs()
@@ -323,6 +378,123 @@ class User(DBUser, AbstractUser, BaseUser):
             finally:
                 self._is_backfilling = False
 
+    @property
+    def _takeout_options(self) -> dict[str, bool | int]:
+        return {
+            "users": True,
+            "chats": self.config["bridge.backfill.normal_groups"],
+            "megagroups": True,
+            "channels": True,
+            "files": True,
+            "max_file_size": min(self.bridge.matrix.media_config.upload_size, 2000 * 1024 * 1024),
+        }
+
+    async def _try_handle_backfill_requests_loop(self) -> None:
+        if not self.config["bridge.backfill.enable"]:
+            return
+        try:
+            await self._handle_backfill_requests_loop()
+        except Exception:
+            self.log.exception("Fatal error in backfill request loop")
+
+    async def _handle_backfill_requests_loop(self) -> None:
+        while True:
+            req = await Backfill.get_next(self.mxid)
+            if not req:
+                try:
+                    await asyncio.wait_for(self.wakeup_backfill_task.wait(), timeout=300)
+                except asyncio.TimeoutError:
+                    pass
+                self.wakeup_backfill_task.clear()
+            else:
+                try:
+                    await self._takeout_and_backfill(req)
+                except Exception:
+                    self.log.exception("Error in takeout backfill loop, retrying in an hour")
+                    await asyncio.sleep(3600)
+
+    async def _check_server_notice_edit(self, message: Message) -> None:
+        if "Data export request" in message.message and "Accepted" in message.message:
+            self.log.debug(
+                f"Received an edit to message {message.id} that looks like the data export"
+                " was accepted, marking takeout as retriable"
+            )
+            self.takeout_retry_immediate.set()
+
+    async def _takeout_and_backfill(self, first_req: Backfill, first_attempt: bool = True) -> None:
+        self.takeout_retry_immediate.clear()
+        self.takeout_requested = True
+        try:
+            async with self.client.takeout(**self._takeout_options) as takeout_client:
+                self.takeout_requested = False
+                self.log.info("Acquired takeout client successfully")
+                await self._backfill_loop_with_client(takeout_client, first_req)
+                self.log.info("Backfills finished, exiting takeout")
+        except TakeoutInitDelayError as e:
+            if first_attempt:
+                self.log.info(
+                    f"Takeout requested, will wait for retry request or {e.seconds} seconds"
+                )
+            else:
+                self.log.warning(
+                    f"Got takeout init delay again after retry, waiting for {e.seconds} seconds"
+                )
+            try:
+                await asyncio.wait_for(self.takeout_retry_immediate.wait(), timeout=e.seconds)
+                self.log.info("Retrying takeout")
+            except asyncio.TimeoutError:
+                self.log.info("Takeout timeout expired")
+            await self._takeout_and_backfill(first_req, first_attempt=False)
+
+    async def _backfill_loop_with_client(
+        self, client: MautrixTelegramClient, first_req: Backfill
+    ) -> None:
+        missed_reqs = 0
+        while missed_reqs < 10:
+            req = first_req or await Backfill.get_next(self.mxid)
+            first_req = None
+            if not req:
+                missed_reqs += 1
+                try:
+                    await asyncio.wait_for(self.wakeup_backfill_task.wait(), timeout=30)
+                except asyncio.TimeoutError:
+                    pass
+                self.wakeup_backfill_task.clear()
+                continue
+            missed_reqs = 0
+            self.log.info("Backfill request %s", req)
+            try:
+                portal = await po.Portal.get_by_tgid(
+                    TelegramID(req.portal_tgid), tg_receiver=TelegramID(req.portal_tg_receiver)
+                )
+                await req.mark_dispatched()
+                if req.type == BackfillType.HISTORICAL:
+                    await portal.backfill(self, client, req=req)
+                elif req.type == BackfillType.SYNC_DIALOG:
+                    await self._backfill_sync_dialog(portal, client, req.extra_data)
+                await req.mark_done()
+                await asyncio.sleep(req.post_batch_delay)
+            except Exception:
+                self.log.exception("Error handling backfill request for %s", req.portal_tgid)
+                await req.set_cooldown_timeout(1800)
+
+    async def _backfill_sync_dialog(
+        self, portal: po.Portal, client: MautrixTelegramClient, post_sync_args: dict[str, Any]
+    ) -> None:
+        if portal.mxid:
+            self.log.debug("Portal already exists, skipping dialog sync backfill queue item")
+            return
+        self.log.info(f"Creating portal for {portal.tgid_log} as part of backfill loop")
+        try:
+            await portal.create_matrix_room(
+                self, client=client, update_if_exists=False, invites=[self.mxid]
+            )
+        except Exception:
+            self.log.exception(f"Error while creating {portal.tgid_log}")
+        else:
+            puppet = await pu.Puppet.get_by_custom_mxid(self.mxid)
+            await self._post_sync_dialog(portal, puppet, was_created=True, **post_sync_args)
+
     async def update(self, update: TypeUpdate) -> bool:
         if not self.is_bot:
             return False
@@ -405,27 +577,46 @@ class User(DBUser, AbstractUser, BaseUser):
                 except MatrixRequestError:
                     pass
 
-    async def log_out(self) -> bool:
+    async def log_out(
+        self,
+        delete: bool = True,
+        do_logout: bool = True,
+        state: BridgeStateEvent = BridgeStateEvent.LOGGED_OUT,
+        error: str | None = None,
+        message: str | None = None,
+    ) -> bool:
         puppet = await pu.Puppet.get_by_tgid(self.tgid)
-        if puppet.is_real_user:
+        if puppet is not None and puppet.is_real_user:
             await puppet.switch_mxid(None, None)
         try:
             await self.kick_from_portals()
         except Exception:
             self.log.exception("Failed to kick user from portals on logout")
-        await self.push_bridge_state(BridgeStateEvent.LOGGED_OUT)
         if self.tgid:
             try:
                 del self.by_tgid[self.tgid]
             except KeyError:
                 pass
-            self.tgid = None
-        ok = await self.client.log_out()
-        sess = self.client.session
-        await self.stop()
-        await sess.delete()
-        await self.delete()
-        self.by_mxid.pop(self.mxid, None)
+        ok = False
+        if self.client is not None:
+            sess = self.client.session
+            # Try to send a logout request. If it succeeds, this also disconnects the client and
+            # deletes the session, but we do those again later just to be safe.
+            if do_logout:
+                ok = await self.client.log_out()
+            # Force-disconnect the client and set it to None
+            await self.stop()
+            await sess.delete()
+
+        # TODO send a management room notice for non-manual logouts?
+        await self.push_bridge_state(state, error=error, message=message)
+        if delete:
+            await self.delete()
+            self.by_mxid.pop(self.mxid, None)
+            self.log.info("User deleted")
+        else:
+            await self.remove_tgid()
+            self.log.info("User telegram ID cleared")
         self._track_metric(METRIC_LOGGED_IN, False)
         return ok
 
@@ -483,19 +674,18 @@ class User(DBUser, AbstractUser, BaseUser):
         if active and tag_info is None:
             tag_info = RoomTagInfo(order=0.5)
             tag_info[DOUBLE_PUPPET_SOURCE_KEY] = self.bridge.name
-            self.log.debug("Adding tag {tag} to {portal.mxid}/{portal.tgid}")
+            self.log.debug(f"Adding tag {tag} to {portal.mxid}/{portal.tgid}")
             await puppet.intent.set_room_tag(portal.mxid, tag, tag_info)
         elif (
             not active and tag_info and tag_info.get(DOUBLE_PUPPET_SOURCE_KEY) == self.bridge.name
         ):
-            self.log.debug("Removing tag {tag} from {portal.mxid}/{portal.tgid}")
+            self.log.debug(f"Removing tag {tag} from {portal.mxid}/{portal.tgid}")
             await puppet.intent.remove_room_tag(portal.mxid, tag)
 
-    async def _mute_room(self, puppet: pu.Puppet, portal: po.Portal, mute_until: datetime) -> None:
+    async def _mute_room(self, puppet: pu.Puppet, portal: po.Portal, mute_until: float) -> None:
         if not self.config["bridge.mute_bridging"] or not portal or not portal.mxid:
             return
-        now = datetime.utcnow().replace(tzinfo=timezone.utc)
-        if mute_until is not None and mute_until > now:
+        if mute_until is not None and mute_until > time.time():
             self.log.debug(
                 f"Muting {portal.mxid}/{portal.tgid} (muted until {mute_until} on Telegram)"
             )
@@ -551,15 +741,28 @@ class User(DBUser, AbstractUser, BaseUser):
         portal = await po.Portal.get_by_entity(
             update.peer.peer, tg_receiver=self.tgid, create=False
         )
-        await self._mute_room(puppet, portal, update.notify_settings.mute_until)
+        await self._mute_room(puppet, portal, update.notify_settings.mute_until.timestamp())
 
     async def _sync_dialog(
         self, portal: po.Portal, dialog: Dialog, should_create: bool, puppet: pu.Puppet | None
     ) -> None:
         was_created = False
+        post_sync_args = {
+            "last_message_ts": cast(datetime, dialog.date).timestamp(),
+            "unread_count": dialog.unread_count,
+            "max_read_id": dialog.dialog.read_inbox_max_id,
+            "mute_until": (
+                dialog.dialog.notify_settings.mute_until.timestamp()
+                if dialog.dialog.notify_settings.mute_until
+                else None
+            ),
+            "pinned": dialog.pinned,
+            "archived": dialog.archived,
+        }
         if portal.mxid:
+            self.log.debug(f"Backfilling and updating {portal.tgid_log} (dialog sync)")
             try:
-                await portal.backfill(self, last_id=dialog.message.id)
+                await portal.forward_backfill(self, initial=False, last_tgid=dialog.message.id)
             except Exception:
                 self.log.exception(f"Error while backfilling {portal.tgid_log}")
             try:
@@ -567,34 +770,73 @@ class User(DBUser, AbstractUser, BaseUser):
             except Exception:
                 self.log.exception(f"Error while updating {portal.tgid_log}")
         elif should_create:
+            self.log.debug(f"Creating portal for {portal.tgid_log} immediately (dialog sync)")
             try:
                 await portal.create_matrix_room(self, dialog.entity, invites=[self.mxid])
                 was_created = True
             except Exception:
                 self.log.exception(f"Error while creating {portal.tgid_log}")
+        elif self.config["bridge.sync_deferred_create_all"]:
+            self.log.debug(f"Enqueuing deferred dialog sync for {portal.tgid_log}")
+            await portal.enqueue_backfill(
+                self,
+                priority=40,
+                type=BackfillType.SYNC_DIALOG,
+                extra_data=post_sync_args,
+            )
         if portal.mxid and puppet and puppet.is_real_user:
-            tg_space = portal.tgid if portal.peer_type == "channel" else self.tgid
-            if dialog.unread_count == 0:
-                # This is usually more reliable than finding a specific message
-                # e.g. if the last read message is a service message that isn't in the message db
-                last_read = await DBMessage.find_last(portal.mxid, tg_space)
-            else:
-                last_read = await DBMessage.get_one_by_tgid(
-                    portal.tgid, tg_space, dialog.dialog.read_inbox_max_id
+            await self._post_sync_dialog(
+                portal=portal,
+                puppet=puppet,
+                was_created=was_created,
+                **post_sync_args,
+            )
+        self.log.debug(f"_sync_dialog finished for {portal.tgid_log}")
+
+    async def _post_sync_dialog(
+        self,
+        portal: po.Portal,
+        puppet: pu.Puppet,
+        was_created: bool,
+        max_read_id: int,
+        last_message_ts: float,
+        unread_count: int,
+        mute_until: float,
+        pinned: bool,
+        archived: bool,
+    ) -> None:
+        self.log.debug(
+            f"Running dialog post-sync for {portal.tgid_log} with args "
+            f"{was_created=}, {max_read_id=}, {last_message_ts=}, {unread_count=}, "
+            f"{mute_until=}, {pinned=}, {archived=}"
+        )
+        tg_space = portal.tgid if portal.peer_type == "channel" else self.tgid
+        unread_threshold_hours = self.config["bridge.backfill.unread_hours_threshold"]
+        force_read = (
+            was_created
+            and unread_threshold_hours >= 0
+            and last_message_ts + (unread_threshold_hours * 60 * 60) < time.time()
+        )
+        if unread_count == 0 or force_read:
+            # This is usually more reliable than finding a specific message
+            # e.g. if the last read message is a service message that isn't in the message db
+            last_read = await DBMessage.find_last(portal.mxid, tg_space)
+            if force_read:
+                self.log.debug(
+                    f"Marking {portal.tgid_log} as read because the last message is from "
+                    f"{last_message_ts} (unread threshold is {unread_threshold_hours} hours)"
                 )
-            try:
-                if last_read:
-                    await puppet.intent.mark_read(last_read.mx_room, last_read.mxid)
-                if was_created or not self.config["bridge.tag_only_on_create"]:
-                    await self._mute_room(puppet, portal, dialog.dialog.notify_settings.mute_until)
-                    await self._tag_room(
-                        puppet, portal, self.config["bridge.pinned_tag"], dialog.pinned
-                    )
-                    await self._tag_room(
-                        puppet, portal, self.config["bridge.archive_tag"], dialog.archived
-                    )
-            except Exception:
-                self.log.exception(f"Error updating read status and tags for {portal.tgid_log}")
+        else:
+            last_read = await DBMessage.get_one_by_tgid(portal.tgid, tg_space, max_read_id)
+        try:
+            if last_read:
+                await puppet.intent.mark_read(last_read.mx_room, last_read.mxid)
+            if was_created or not self.config["bridge.tag_only_on_create"]:
+                await self._mute_room(puppet, portal, mute_until)
+                await self._tag_room(puppet, portal, self.config["bridge.pinned_tag"], pinned)
+                await self._tag_room(puppet, portal, self.config["bridge.archive_tag"], archived)
+        except Exception:
+            self.log.exception(f"Error updating read status and tags for {portal.tgid_log}")
 
     async def get_cached_portals(self) -> dict[tuple[TelegramID, TelegramID], po.Portal]:
         if self._portals_cache is None:
@@ -611,9 +853,7 @@ class User(DBUser, AbstractUser, BaseUser):
         update_limit = self.config["bridge.sync_update_limit"] or None
         create_limit = self.config["bridge.sync_create_limit"]
         index = 0
-        self.log.debug(
-            f"Syncing dialogs (update_limit={update_limit}, create_limit={create_limit})"
-        )
+        self.log.debug(f"Syncing dialogs ({update_limit=}, {create_limit=})")
         await self.push_bridge_state(BridgeStateEvent.BACKFILLING)
         puppet = await pu.Puppet.get_by_custom_mxid(self.mxid)
         dialog: Dialog
@@ -634,11 +874,12 @@ class User(DBUser, AbstractUser, BaseUser):
                 continue
             portal = await po.Portal.get_by_entity(entity, tg_receiver=self.tgid)
             new_portal_cache[portal.tgid_full] = portal
+            should_create = not create_limit or index < create_limit
             coro = self._sync_dialog(
                 portal=portal,
                 dialog=dialog,
                 puppet=puppet,
-                should_create=not create_limit or index < create_limit,
+                should_create=should_create,
             )
             creators.append(asyncio.create_task(coro))
             index += 1
@@ -700,6 +941,48 @@ class User(DBUser, AbstractUser, BaseUser):
         self.log.debug("Contact syncing complete")
         return contacts
 
+    async def get_available_reactions(self) -> set[str]:
+        if self._available_emoji_reactions_fetched + 12 * 60 * 60 > time.monotonic():
+            return self._available_emoji_reactions
+        async with self._available_emoji_reactions_lock:
+            if self._available_emoji_reactions_fetched + 12 * 60 * 60 > time.monotonic():
+                return self._available_emoji_reactions
+            self.log.debug("Fetching available emoji reactions")
+            available_reactions = await self.client(
+                GetAvailableReactionsRequest(hash=self._available_emoji_reactions_hash or 0)
+            )
+            if isinstance(available_reactions, AvailableReactions):
+                self._available_emoji_reactions = {
+                    react.reaction
+                    for react in available_reactions.reactions
+                    if self.is_premium or not react.premium
+                }
+                self._available_emoji_reactions_hash = available_reactions.hash
+                self._available_emoji_reactions_fetched = time.monotonic()
+                self.log.debug(
+                    "Got available emoji reactions: %s", self._available_emoji_reactions
+                )
+            return self._available_emoji_reactions
+
+    def tl_to_json(self) -> Any:
+        pass
+
+    async def get_app_config(self) -> dict[str, Any]:
+        if not self._app_config:
+            cfg = await self.client(GetAppConfigRequest())
+            self._app_config = util.parse_tl_json(cfg)
+        return self._app_config
+
+    async def get_max_reactions(self, is_premium: bool | None = None) -> int:
+        if is_premium is None:
+            is_premium = self.is_premium
+        cfg = await self.get_app_config()
+        return (
+            cfg.get("reactions_user_max_premium", 3)
+            if is_premium
+            else cfg.get("reactions_user_max_default", 1)
+        )
+
     # endregion
     # region Class instance lookup
 

mautrix_telegram/util/__init__.py

@@ -1,4 +1,11 @@
 from .color_log import ColorFormatter
-from .file_transfer import convert_image, transfer_custom_emojis_to_matrix, transfer_file_to_matrix
+from .file_transfer import (
+    UnicodeCustomEmoji,
+    convert_image,
+    transfer_custom_emojis_to_matrix,
+    transfer_file_to_matrix,
+    unicode_custom_emoji_map,
+)
 from .parallel_file_transfer import parallel_transfer_to_telegram
 from .recursive_dict import recursive_del, recursive_get, recursive_set
+from .tl_json import parse_tl_json

mautrix_telegram/util/file_transfer.py

@@ -15,12 +15,13 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 from __future__ import annotations
 
-from typing import Optional, Union
+from typing import NamedTuple, Optional, Union
 from io import BytesIO
 from sqlite3 import IntegrityError
 import asyncio
 import logging
-import tempfile
+import pickle
+import pkgutil
 import time
 
 from asyncpg import UniqueViolationError
@@ -44,7 +45,7 @@ from telethon.tl.types import (
 )
 
 from mautrix.appservice import IntentAPI
-from mautrix.util import magic
+from mautrix.util import ffmpeg, magic, variation_selector
 
 from .. import abstract_user as au
 from ..db import TelegramFile as DBTelegramFile
@@ -59,11 +60,6 @@ try:
 except ImportError:
     Image = None
 
-try:
-    from moviepy.editor import VideoFileClip
-except ImportError:
-    VideoFileClip = None
-
 try:
     from mautrix.crypto.attachments import encrypt_attachment
 except ImportError:
@@ -101,29 +97,16 @@ def convert_image(
         return source_mime, file, None, None
 
 
-def _read_video_thumbnail(
-    data: bytes,
-    video_ext: str = "mp4",
-    frame_ext: str = "png",
-    max_size: tuple[int, int] = (1024, 720),
-) -> tuple[bytes, int, int]:
-    with tempfile.NamedTemporaryFile(prefix="mxtg_video_", suffix=f".{video_ext}") as file:
-        # We don't have any way to read the video from memory, so save it to disk.
-        file.write(data)
-
-        # Read temp file and get frame
-        frame = VideoFileClip(file.name).get_frame(0)
-
-    # Convert to png and save to BytesIO
-    image = Image.fromarray(frame).convert("RGBA")
-
-    thumbnail_file = BytesIO()
-    if max_size:
-        image.thumbnail(max_size, Image.ANTIALIAS)
-    image.save(thumbnail_file, frame_ext)
-
-    w, h = image.size
-    return thumbnail_file.getvalue(), w, h
+async def _read_video_thumbnail(data: bytes, mime_type: str) -> tuple[bytes, int, int]:
+    first_frame = await ffmpeg.convert_bytes(
+        data,
+        output_extension=".png",
+        output_args=("-update", "1", "-frames:v", "1"),
+        input_mime=mime_type,
+        logger=log,
+    )
+    width, height = Image.open(BytesIO(first_frame)).size
+    return first_frame, width, height
 
 
 def _location_to_id(location: TypeLocation) -> str:
@@ -149,7 +132,7 @@ async def transfer_thumbnail_to_matrix(
     height: int | None = None,
     async_upload: bool = False,
 ) -> DBTelegramFile | None:
-    if not Image or not VideoFileClip:
+    if not Image or not ffmpeg.ffmpeg_path:
         return None
 
     loc_id = _location_to_id(thumbnail_loc)
@@ -168,10 +151,12 @@ async def transfer_thumbnail_to_matrix(
     video_ext = sane_mimetypes.guess_extension(mime_type)
     if custom_data:
         file = custom_data
-    elif VideoFileClip and video_ext and video:
+    elif video_ext and video:
+        log.debug(f"Generating thumbnail for video {loc_id} with ffmpeg")
         try:
-            file, width, height = _read_video_thumbnail(video, video_ext, frame_ext="png")
-        except OSError:
+            file, width, height = await _read_video_thumbnail(video, mime_type=mime_type)
+        except Exception:
+            log.warning(f"Failed to generate thumbnail for {loc_id}", exc_info=True)
             return None
         mime_type = "image/png"
     else:
@@ -212,20 +197,44 @@ async def transfer_thumbnail_to_matrix(
 
 transfer_locks: dict[str, asyncio.Lock] = {}
 
+unicode_custom_emoji_map = pickle.loads(
+    pkgutil.get_data("mautrix_telegram", "unicodemojipack.pickle")
+)
+reverse_unicode_custom_emoji_map = {
+    doc_id: emoji for emoji, doc_id in unicode_custom_emoji_map.items()
+}
+
 TypeThumbnail = Optional[Union[TypeLocation, TypePhotoSize]]
 
 
+class UnicodeCustomEmoji(NamedTuple):
+    emoji: str
+
+
 async def transfer_custom_emojis_to_matrix(
-    source: au.AbstractUser, emoji_ids: list[int]
-) -> dict[int, DBTelegramFile]:
+    source: au.AbstractUser, emoji_ids: list[int], client: MautrixTelegramClient | None = None
+) -> dict[int, DBTelegramFile | UnicodeCustomEmoji]:
+    if not client:
+        client = source.client
     emoji_ids = set(emoji_ids)
+    existing_unicode = {}
+    for emoji_id in emoji_ids:
+        try:
+            existing_unicode[emoji_id] = UnicodeCustomEmoji(
+                variation_selector.add(reverse_unicode_custom_emoji_map[emoji_id])
+            )
+        except KeyError:
+            pass
+    emoji_ids -= existing_unicode.keys()
+    if not emoji_ids:
+        return existing_unicode
     existing = await DBTelegramFile.get_many([str(id) for id in emoji_ids])
-    file_map = {int(file.id): file for file in existing}
+    file_map = {int(file.id): file for file in existing} | existing_unicode
     not_existing_ids = list(emoji_ids - file_map.keys())
     if not_existing_ids:
         log.debug(f"Transferring custom emojis through {source.mxid}: {not_existing_ids}")
 
-        documents: list[Document] = await source.client(
+        documents: list[Document] = await client(
             GetCustomEmojiDocumentsRequest(document_id=not_existing_ids)
         )
 
@@ -237,7 +246,7 @@ async def transfer_custom_emojis_to_matrix(
         async def transfer(document: Document) -> None:
             async with transfer_sema:
                 file_map[document.id] = await transfer_file_to_matrix(
-                    source.client,
+                    client,
                     source.bridge.az.intent,
                     document,
                     is_sticker=True,
@@ -324,10 +333,10 @@ async def _unlocked_transfer_file_to_matrix(
             client, intent, loc_id, location, filename, encrypt, parallel_id
         )
         mime_type = location.mime_type
-        file = None
+        unencrypted_file = None
     else:
         try:
-            file = await client.download_file(location)
+            unencrypted_file = file = await client.download_file(location)
         except (LocationInvalidError, FileIdInvalidError):
             return None
         except (AuthBytesInvalidError, AuthKeyInvalidError, SecurityError) as e:
@@ -375,34 +384,37 @@ async def _unlocked_transfer_file_to_matrix(
             width=width,
             height=height,
         )
-    if thumbnail and (mime_type.startswith("video/") or mime_type == "image/gif"):
-        if isinstance(thumbnail, (PhotoSize, PhotoCachedSize)):
-            thumbnail = thumbnail.location
-        try:
+    try:
+        if thumbnail and (mime_type.startswith("video/") or mime_type == "image/gif"):
+            if isinstance(thumbnail, (PhotoSize, PhotoCachedSize)):
+                thumbnail = thumbnail.location
+            try:
+                db_file.thumbnail = await transfer_thumbnail_to_matrix(
+                    client,
+                    intent,
+                    thumbnail,
+                    video=unencrypted_file,
+                    mime_type=mime_type,
+                    encrypt=encrypt,
+                    async_upload=async_upload,
+                )
+            except FileIdInvalidError:
+                log.warning(f"Failed to transfer thumbnail {thumbnail!s}", exc_info=True)
+        elif converted_anim and converted_anim.thumbnail_data:
             db_file.thumbnail = await transfer_thumbnail_to_matrix(
                 client,
                 intent,
-                thumbnail,
-                video=file,
-                mime_type=mime_type,
+                location,
+                video=None,
                 encrypt=encrypt,
+                custom_data=converted_anim.thumbnail_data,
+                mime_type=converted_anim.thumbnail_mime,
+                width=converted_anim.width,
+                height=converted_anim.height,
                 async_upload=async_upload,
             )
-        except FileIdInvalidError:
-            log.warning(f"Failed to transfer thumbnail for {thumbnail!s}", exc_info=True)
-    elif converted_anim and converted_anim.thumbnail_data:
-        db_file.thumbnail = await transfer_thumbnail_to_matrix(
-            client,
-            intent,
-            location,
-            video=None,
-            encrypt=encrypt,
-            custom_data=converted_anim.thumbnail_data,
-            mime_type=converted_anim.thumbnail_mime,
-            width=converted_anim.width,
-            height=converted_anim.height,
-            async_upload=async_upload,
-        )
+    except Exception:
+        log.exception(f"Failed to transfer thumbnail for {loc_id}")
 
     try:
         await db_file.insert()

mautrix_telegram/util/tl_json.py

@@ -0,0 +1,39 @@
+# mautrix-telegram - A Matrix-Telegram puppeting bridge
+# Copyright (C) 2022 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from telethon.tl.types import (
+    JsonArray,
+    JsonBool,
+    JsonNull,
+    JsonNumber,
+    JsonObject,
+    JsonObjectValue,
+    JsonString,
+    TypeJSONValue,
+)
+
+from mautrix.types import JSON
+
+
+def parse_tl_json(val: TypeJSONValue) -> JSON:
+    if isinstance(val, JsonObject):
+        return {entry.key: parse_tl_json(entry.value) for entry in val.value}
+    elif isinstance(val, JsonArray):
+        return [parse_tl_json(item) for item in val.value]
+    elif isinstance(val, (JsonBool, JsonNumber, JsonString)):
+        return val.value
+    elif isinstance(val, JsonNull):
+        return None
+    raise ValueError(f"Unsupported type {type(val)} in TL JSON object")

mautrix_telegram/web/common/auth_api.py

@@ -38,6 +38,7 @@ from telethon.errors import (
 )
 
 from mautrix.bridge import InvalidAccessToken, OnlyLoginSelf
+from mautrix.util import background_task
 from mautrix.util.format_duration import format_duration
 
 from ...commands.telegram.auth import enter_password
@@ -128,7 +129,7 @@ class AuthAPI(abc.ABC):
                 status=200,
                 message=(
                     "Code requested successfully. Check your SMS "
-                    "or Telegram client and enter the code below."
+                    "or Telegram app and enter the code below."
                 ),
             )
         except PhoneNumberInvalidError:
@@ -199,7 +200,7 @@ class AuthAPI(abc.ABC):
         existing_user = await User.get_by_tgid(user_info.id)
         if existing_user and existing_user != user:
             await existing_user.log_out()
-        asyncio.create_task(user.post_login(user_info, first_login=True))
+        background_task.create(user.post_login(user_info, first_login=True))
         if user.command_status and user.command_status["action"] == "Login":
             user.command_status = None
 
@@ -341,8 +342,16 @@ class AuthAPI(abc.ABC):
                 errcode="password_invalid",
                 error="Incorrect password.",
             )
-        except Exception:
+        except Exception as e:
             self.log.exception("Error sending password")
+            if isinstance(e, ValueError) and "You must provide a phone and a code" in str(e):
+                return self.get_login_response(
+                    mxid=user.mxid,
+                    state="request",
+                    status=400,
+                    errcode="phone_code_not_entered",
+                    error="Please request a new phone code and enter it first.",
+                )
             return self.get_login_response(
                 mxid=user.mxid,
                 state="password",

mautrix_telegram/web/provisioning/__init__.py

@@ -17,16 +17,22 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING, Awaitable, Callable
 import asyncio
+import datetime
 import json
 import logging
 
 from aiohttp import web
+from telethon.errors import SessionPasswordNeededError
+from telethon.tl.custom import QRLogin
+from telethon.tl.functions.messages import GetAllStickersRequest
 from telethon.tl.types import ChannelForbidden, ChatForbidden, TypeChat, User as TLUser
 from telethon.utils import get_peer_id, resolve_id
 
 from mautrix.appservice import AppService
+from mautrix.client import Client
 from mautrix.errors import IntentError, MatrixRequestError
 from mautrix.types import UserID
+from mautrix.util import background_task
 
 from ...commands.portal.util import get_initial_state, user_has_power_level
 from ...portal import Portal
@@ -53,7 +59,7 @@ class ProvisioningAPI(AuthAPI):
 
         self.app = web.Application(loop=bridge.loop, middlewares=[self.error_middleware])
 
-        portal_prefix = "/v1/portal/{mxid:![^/]+}"
+        portal_prefix = "/v1/portal/{mxid}"
         self.app.router.add_route("GET", f"{portal_prefix}", self.get_portal_by_mxid)
         self.app.router.add_route("GET", "/v1/portal/{tgid:-[0-9]+}", self.get_portal_by_tgid)
         self.app.router.add_route(
@@ -62,7 +68,7 @@ class ProvisioningAPI(AuthAPI):
         self.app.router.add_route("POST", f"{portal_prefix}/create", self.create_chat)
         self.app.router.add_route("POST", f"{portal_prefix}/disconnect", self.disconnect_chat)
 
-        user_prefix = "/v1/user/{mxid:@[^:]*:[^/]+}"
+        user_prefix = "/v1/user/{mxid}"
         self.app.router.add_route("GET", f"{user_prefix}", self.get_user_info)
         self.app.router.add_route("GET", f"{user_prefix}/chats", self.get_chats)
         self.app.router.add_route("GET", f"{user_prefix}/contacts", self.get_contacts)
@@ -71,7 +77,12 @@ class ProvisioningAPI(AuthAPI):
         )
         self.app.router.add_route("POST", f"{user_prefix}/pm/{{identifier}}", self.start_dm)
 
+        self.app.router.add_route("GET", f"{user_prefix}/stickersets", self.get_stickersets)
+
+        self.app.router.add_route("POST", f"{user_prefix}/retry_takeout", self.retry_takeout)
+
         self.app.router.add_route("POST", f"{user_prefix}/logout", self.logout)
+        self.app.router.add_route("GET", f"{user_prefix}/login/qr", self.login_qr)
         self.app.router.add_route("POST", f"{user_prefix}/login/bot_token", self.send_bot_token)
         self.app.router.add_route("POST", f"{user_prefix}/login/request_code", self.request_code)
         self.app.router.add_route("POST", f"{user_prefix}/login/send_code", self.send_code)
@@ -217,7 +228,7 @@ class ProvisioningAPI(AuthAPI):
             portal.photo_id = ""
             await portal.save()
 
-        asyncio.create_task(portal.update_matrix_room(user, entity, levels=levels))
+        background_task.create(portal.update_matrix_room(user, entity, levels=levels))
 
         return web.Response(status=202, body="{}")
 
@@ -338,7 +349,7 @@ class ProvisioningAPI(AuthAPI):
                 self.log.exception("Failed to disconnect chat")
                 return self.get_error_response(500, "exception", "Failed to disconnect chat")
         else:
-            asyncio.create_task(coro)
+            background_task.create(coro)
         return web.json_response({}, status=200 if sync else 202)
 
     async def get_user_info(self, request: web.Request) -> web.Response:
@@ -494,6 +505,78 @@ class ProvisioningAPI(AuthAPI):
             status=201 if just_created else 200,
         )
 
+    async def get_stickersets(self, request: web.Request) -> web.Response:
+        _, user, err = await self.get_user_request_info(
+            request, expect_logged_in=True, want_data=False
+        )
+        if err is not None:
+            return err
+        result = await user.client(GetAllStickersRequest(0))
+        resp = []
+        for stickerset in result.sets:
+            resp.append(stickerset.short_name)
+        return web.json_response(resp, status=200)
+
+    async def retry_takeout(self, request: web.Request) -> web.Response:
+        data, user, err = await self.get_user_request_info(
+            request, expect_logged_in=True, want_data=False
+        )
+        if err is not None:
+            return err
+        if not user.takeout_requested:
+            return web.json_response(
+                {
+                    "error": "There was no takeout requested",
+                },
+                status=400,
+            )
+        user.takeout_retry_immediate.set()
+        return web.json_response({}, status=200)
+
+    async def login_qr(self, request: web.Request) -> web.Response:
+        _, user, err = await self.get_user_request_info(request, websocket=True)
+        if err is not None:
+            return err
+
+        await user.ensure_started(even_if_no_session=True)
+        qr_login = QRLogin(user.client, ignored_ids=[])
+
+        ws = web.WebSocketResponse(protocols=["net.maunium.telegram.login"])
+        await ws.prepare(request)
+
+        retries = 0
+        user_info = None
+        while retries < 4:
+            try:
+                await qr_login.recreate()
+                await ws.send_json(
+                    {
+                        "code": qr_login.url,
+                        "timeout": int(
+                            (
+                                qr_login.expires - datetime.datetime.now(tz=datetime.timezone.utc)
+                            ).total_seconds()
+                        ),
+                    }
+                )
+                user_info = await qr_login.wait()
+                break
+            except asyncio.TimeoutError:
+                retries += 1
+            except SessionPasswordNeededError:
+                await ws.send_json({"success": False, "error": "password-needed"})
+                await ws.close()
+                return ws
+        else:
+            await ws.send_json({"success": False, "error": "timeout"})
+            await ws.close()
+            return ws
+
+        await self.postprocess_login(user, user_info)
+        await ws.send_json({"success": True})
+        await ws.close()
+        return ws
+
     async def send_bot_token(self, request: web.Request) -> web.Response:
         data, user, err = await self.get_user_request_info(request)
         if err is not None:
@@ -619,6 +702,15 @@ class ProvisioningAPI(AuthAPI):
             )
         return None
 
+    def check_websocket_authorization(self, request: web.Request) -> web.Response | None:
+        auth_parts = request.headers.get("Sec-WebSocket-Protocol").split(",")
+        for part in auth_parts:
+            if part.strip() == f"net.maunium.telegram.auth-{self.secret}":
+                return None
+        return self.get_error_response(
+            error="Shared secret is not valid.", errcode="shared_secret_invalid", status=401
+        )
+
     @staticmethod
     async def get_data(request: web.Request) -> dict | None:
         try:
@@ -639,6 +731,12 @@ class ProvisioningAPI(AuthAPI):
             return None, self.get_login_response(
                 error="User ID not given.", errcode="mxid_empty", status=400
             )
+        try:
+            Client.parse_user_id(mxid)
+        except ValueError:
+            return None, self.get_login_response(
+                error="Invalid user ID", errcode="mxid_invalid", status=400
+            )
 
         user = await User.get_and_start_by_mxid(mxid, even_if_no_session=True)
         if require_puppeting and not user.puppet_whitelisted:
@@ -667,8 +765,12 @@ class ProvisioningAPI(AuthAPI):
         expect_logged_in: bool | None = False,
         require_puppeting: bool = False,
         want_data: bool = True,
+        websocket: bool = False,
     ) -> tuple[dict | None, User | None, web.Response | None]:
-        err = self.check_authorization(request)
+        if not websocket:
+            err = self.check_authorization(request)
+        else:
+            err = self.check_websocket_authorization(request)
         if err is not None:
             return None, None, err
 

optional-requirements.txt

@@ -2,8 +2,7 @@
 # Uncommented lines after the group definition insert things into that group.
 
 #/speedups
-cryptg>=0.1,<0.4
-cchardet
+cryptg>=0.1,<0.5
 aiodns
 brotli
 
@@ -11,14 +10,12 @@ brotli
 pillow>=4,<10
 qrcode>=6,<8
 
-#/hq_thumbnails
-moviepy>=1,<2
 
 #/formattednumbers
 phonenumbers>=8,<9
 
 #/metrics
-prometheus_client>=0.6,<0.15
+prometheus_client>=0.6,<0.17
 
 #/e2be
 python-olm>=3,<4
@@ -26,4 +23,4 @@ pycryptodome>=3,<4
 unpaddedbase64>=1,<3
 
 #/sqlite
-aiosqlite>=0.16,<0.18
+aiosqlite>=0.16,<0.19

requirements.txt

@@ -3,9 +3,9 @@ python-magic>=0.4,<0.5
 commonmark>=0.8,<0.10
 aiohttp>=3,<4
 yarl>=1,<2
-mautrix>=0.18.2,<0.19
-#telethon>=1.24,<1.25
-tulir-telethon==1.26.0a5
-asyncpg>=0.20,<0.27
+mautrix>=0.19.4,<0.20
+#telethon>=1.25.4,<1.27
+tulir-telethon==1.28.0a3
+asyncpg>=0.20,<0.28
 mako>=1,<2
 setuptools

setup.py

@@ -66,7 +66,7 @@ setuptools.setup(
     ],
     package_data={"mautrix_telegram": [
         "web/public/*.mako", "web/public/*.png", "web/public/*.css",
-        "example-config.yaml",
+        "example-config.yaml", "unicodemojipack.pickle",
     ]},
     data_files=[
         (".", ["mautrix_telegram/example-config.yaml"]),