Bump version to 0.12.2

Fixes #867

.gitignore

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

CHANGELOG.md

@@ -1,3 +1,55 @@
+# 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
+* Support for custom emojis in reactions.
+  * Like other bridges with custom emoji reactions, they're bridged as `mxc://`
+    URIs, so client support is required to render them properly.
+
+### Improved
+* The bridge will now poll for reactions to 20 most recent messages when
+  receiving a read receipt. This works around Telegram's bad protocol that
+  doesn't notify clients on reactions to other users' messages.
+* The docker image now has an option to bypass the startup script by setting
+  the `MAUTRIX_DIRECT_STARTUP` environment variable. Additionally, it will
+  refuse to run as a non-root user if that variable is not set (and print an
+  error message suggesting to either set the variable or use a custom command).
+* Moved environment variable overrides for config fields to mautrix-python.
+  The new system also allows loading JSON values to enable overriding maps like
+  `login_shared_secret_map`.
+
+### Fixed
+* `ChatParticipantsForbidden` is handled properly when syncing non-supergroup
+  info.
+* Fixed some bugs with file transfers when using SQLite.
+* Fixed error when attempting to log in again after logging out.
+* Fixed QR login not working.
+* Fixed error syncing chats if bridging a message had previously been
+  interrupted.
+
 # v0.12.0 (2022-08-26)
 
 **N.B.** This release requires a homeserver with Matrix v1.1 support, which

Dockerfile

@@ -1,7 +1,5 @@
 FROM dock.mau.dev/tulir/lottieconverter:alpine-3.16
 
-ARG TARGETARCH=amd64
-
 RUN apk add --no-cache \
       python3 py3-pip py3-setuptools py3-wheel \
       py3-pillow \

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:

docker-run.sh

@@ -1,4 +1,18 @@
 #!/bin/sh
+if [ ! -z "$MAUTRIX_DIRECT_STARTUP" ]; then
+	if [ $(id -u) == 0 ]; then
+		echo "|------------------------------------------|"
+		echo "| Warning: running bridge unsafely as root |"
+		echo "|------------------------------------------|"
+	fi
+	exec python3 -m mautrix_telegram -c /data/config.yaml
+elif [ $(id -u) != 0 ]; then
+	echo "The startup script must run as root. It will use su-exec to drop permissions before running the bridge."
+	echo "To bypass the startup script, either set the `MAUTRIX_DIRECT_STARTUP` environment variable,"
+	echo "or just use `python3 -m mautrix_telegram -c /data/config.yaml` as the run command."
+	echo "Note that the config and registration will not be auto-generated when bypassing the startup script."
+	exit 1
+fi
 
 # Define functions.
 function fixperms {

mautrix_telegram/__init__.py

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

mautrix_telegram/abstract_user.py

@@ -22,6 +22,7 @@ import logging
 import platform
 import time
 
+from telethon.errors import UnauthorizedError
 from telethon.network import (
     Connection,
     ConnectionTcpFull,
@@ -40,6 +41,7 @@ from telethon.tl.types import (
     TypeUpdate,
     UpdateChannel,
     UpdateChannelUserTyping,
+    UpdateChatDefaultBannedRights,
     UpdateChatParticipantAdmin,
     UpdateChatParticipants,
     UpdateChatUserTyping,
@@ -238,6 +240,9 @@ class AbstractUser(ABC):
             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")
             await asyncio.sleep(60)
             self.log.debug("Now recreating Telethon update loop")
@@ -297,17 +302,18 @@ class AbstractUser(ABC):
     async def ensure_started(self, even_if_no_session=False) -> AbstractUser:
         if self.connected:
             return self
-        if even_if_no_session or await PgSession.has(self.mxid):
+        session_exists = await PgSession.has(self.mxid)
+        if even_if_no_session or session_exists:
             self.log.debug(
-                "Starting client due to ensure_started"
+                f"Starting client due to ensure_started({even_if_no_session=}, {session_exists=})"
-                f"(even_if_no_session={even_if_no_session})"
             )
             await self.start(delete_unless_authenticated=not even_if_no_session)
         return self
 
     async def stop(self) -> None:
-        await self.client.disconnect()
+        if self.client:
-        self.client = None
+            await self.client.disconnect()
+            self.client = None
 
     # region Telegram update handling
 
@@ -341,6 +347,8 @@ 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)):
@@ -387,6 +395,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)
@@ -415,6 +429,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):
@@ -428,6 +443,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
@@ -435,6 +451,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)
@@ -510,9 +529,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))
+            portal = await po.Portal.get_by_tgid(TelegramID(update.chat_id), peer_type="chat")
-            if not portal:
-                self.log.warning(f"Received message in chat with unknown type {update.chat_id}")
             sender = await pu.Puppet.get_by_tgid(TelegramID(update.from_id))
         elif isinstance(update, UpdateShortMessage):
             portal = await po.Portal.get_by_tgid(
@@ -536,6 +553,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:
@@ -615,7 +634,10 @@ class AbstractUser(ABC):
             self.log.info(
                 "Creating Matrix room 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)
@@ -650,7 +672,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,
@@ -668,6 +690,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,7 +15,7 @@
 # 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
 
@@ -29,6 +29,7 @@ from telethon.tl.types import (
     ChatForbidden,
     ChatParticipantAdmin,
     ChatParticipantCreator,
+    ChatParticipantsForbidden,
     InputChannel,
     InputUser,
     MessageActionChatAddUser,
@@ -56,6 +57,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[
@@ -86,6 +90,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",
@@ -112,6 +117,7 @@ class Bot(AbstractUser):
         )
         self._me_info = None
         self._me_mxid = None
+        self._login_wait_fut = self.loop.create_future()
 
     async def get_me(self, use_cache: bool = True) -> tuple[User, UserID]:
         if not use_cache or not self._me_mxid:
@@ -145,6 +151,9 @@ class Bot(AbstractUser):
         self.tgid = TelegramID(info.id)
         self.tg_username = info.username
         self.mxid = pu.Puppet.get_mxid_from_id(self.tgid)
+        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))
@@ -198,6 +207,8 @@ class Bot(AbstractUser):
             return pcp
         elif isinstance(chat, PeerChat):
             chat = await self.client(GetFullChatRequest(chat.chat_id))
+            if isinstance(chat.full_chat.participants, ChatParticipantsForbidden):
+                return None
             participants = chat.full_chat.participants.participants
             for p in participants:
                 self._admin_cache[chat.channel_id, tgid] = (p, time.time())
@@ -415,6 +426,8 @@ class Bot(AbstractUser):
             await self.add_chat(TelegramID(action.channel_id), "channel")
 
     async def update(self, update) -> bool:
+        if self._login_wait_fut:
+            await self._login_wait_fut
         if not isinstance(update, (UpdateNewMessage, UpdateNewChannelMessage)):
             return False
         if isinstance(update.message, MessageService):

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
@@ -427,6 +428,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 +439,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:
+    if portal.backfill_msc2716:
-        await portal.backfill(evt.sender, limit=limit)
+        messages_per_batch = evt.config["bridge.backfill.incremental.messages_per_batch"]
-    except TakeoutInitDelayError:
+        batches = math.ceil(limit / messages_per_batch)
-        msg = (
+        rounded = ""
-            "Please accept the data export request from a mobile device, "
+        if batches * messages_per_batch != limit:
-            "then re-run the backfill command."
+            rounded = f" (rounded message limit to {batches}*{messages_per_batch})"
-        )
+        await portal.enqueue_backfill(evt.sender, priority=0, max_batches=batches)
-        if portal.peer_type == "user":
+        await evt.reply(f"Backfill queued{rounded}")
-            from mautrix.appservice import IntentAPI
+    else:
-
+        output = await portal.forward_backfill(evt.sender, initial=False, override_limit=limit)
-            await portal.main_intent.send_notice(evt.room_id, msg)
+        await evt.reply(output)
-        else:
-            await evt.reply(msg)

mautrix_telegram/config.py

@@ -35,12 +35,6 @@ Permissions = NamedTuple(
 
 
 class Config(BaseBridgeConfig):
-    def __getitem__(self, key: str) -> Any:
-        try:
-            return os.environ[f"MAUTRIX_TELEGRAM_{key.replace('.', '_').upper()}"]
-        except KeyError:
-            return super().__getitem__(key)
-
     @property
     def forbidden_defaults(self) -> List[ForbiddenDefault]:
         return [
@@ -63,8 +57,6 @@ class Config(BaseBridgeConfig):
         super().do_update(helper)
         copy, copy_dict, base = helper
 
-        copy("homeserver.asmux")
-
         if "appservice.protocol" in self and "appservice.address" not in self:
             protocol, hostname, port = (
                 self["appservice.protocol"],
@@ -121,6 +113,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")
@@ -144,6 +137,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")
@@ -165,12 +159,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.enable")
-        copy("bridge.backfill.takeout_limit")
+        copy("bridge.backfill.msc2716")
-        copy("bridge.backfill.initial_limit")
+        copy("bridge.backfill.double_puppet_backfill")
-        copy("bridge.backfill.missed_limit")
-        copy("bridge.backfill.disable_notifications")
         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 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) -> None:
+        await 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 {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)
@@ -152,6 +161,17 @@ class Message:
             rows = await cls.db.fetch(q, mx_room, tg_space, *mxids)
         return [cls._from_row(row) for row in rows]
 
+    @classmethod
+    async def find_recent(
+        cls, mx_room: RoomID, not_sender: TelegramID, limit: int = 20
+    ) -> list[Message]:
+        q = f"""
+        SELECT {cls.columns} FROM message
+        WHERE mx_room=$1 AND sender<>$2
+        ORDER BY tgid DESC LIMIT $3
+        """
+        return [cls._from_row(row) for row in await cls.db.fetch(q, mx_room, not_sender, limit)]
+
     @classmethod
     async def replace_temp_mxid(cls, temp_mxid: str, mx_room: RoomID, real_mxid: EventID) -> None:
         q = "UPDATE message SET mxid=$1 WHERE mxid=$2 AND mx_room=$3"
@@ -162,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 (
@@ -177,13 +214,7 @@ class Message:
         )
 
     async def insert(self) -> None:
-        q = """
+        await self.db.execute(self._insert_query, *self._values)
-            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)
 
     async def delete(self) -> None:
         q = "DELETE FROM message WHERE mxid=$1 AND mx_room=$2 AND tg_space=$3"

mautrix_telegram/db/puppet.py

@@ -50,6 +50,7 @@ class Puppet:
     avatar_set: bool
     is_bot: bool | None
     is_channel: bool
+    is_premium: bool
 
     custom_mxid: UserID | None
     access_token: str | None
@@ -67,7 +68,8 @@ class Puppet:
     columns: ClassVar[str] = (
         "id, is_registered, displayname, displayname_source, displayname_contact, "
         "displayname_quality, disable_updates, username, phone, photo_id, avatar_url, "
-        "name_set, avatar_set, is_bot, is_channel, custom_mxid, access_token, next_batch, base_url"
+        "name_set, avatar_set, is_bot, is_channel, is_premium, "
+        "custom_mxid, access_token, next_batch, base_url"
     )
 
     @classmethod
@@ -108,6 +110,7 @@ class Puppet:
             self.avatar_set,
             self.is_bot,
             self.is_channel,
+            self.is_premium,
             self.custom_mxid,
             self.access_token,
             self.next_batch,
@@ -120,7 +123,7 @@ class Puppet:
         SET is_registered=$2, displayname=$3, displayname_source=$4, displayname_contact=$5,
             displayname_quality=$6, disable_updates=$7, username=$8, phone=$9, photo_id=$10,
             avatar_url=$11, name_set=$12, avatar_set=$13, is_bot=$14, is_channel=$15,
-            custom_mxid=$16, access_token=$17, next_batch=$18, base_url=$19
+            is_premium=$16, custom_mxid=$17, access_token=$18, next_batch=$19, base_url=$20
         WHERE id=$1
         """
         await self.db.execute(q, *self._values)
@@ -130,8 +133,9 @@ class Puppet:
         INSERT INTO puppet (
             id, is_registered, displayname, displayname_source, displayname_contact,
             displayname_quality, disable_updates, username, phone, photo_id, avatar_url, name_set,
-            avatar_set, is_bot, is_channel, custom_mxid, access_token, next_batch, base_url
+            avatar_set, is_bot, is_channel, is_premium, custom_mxid, access_token, next_batch,
+            base_url
         ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,
-                  $19)
+                  $19, $20)
         """
         await self.db.execute(q, *self._values)

mautrix_telegram/db/reaction.py

@@ -19,6 +19,7 @@ from typing import TYPE_CHECKING, ClassVar
 
 from asyncpg import Record
 from attr import dataclass
+from telethon.tl.types import ReactionCustomEmoji, ReactionEmoji, TypeReaction
 
 from mautrix.types import EventID, RoomID
 from mautrix.util.async_db import Database
@@ -58,9 +59,10 @@ class Reaction:
     @classmethod
     async def get_by_sender(
         cls, mxid: EventID, mx_room: RoomID, tg_sender: TelegramID
-    ) -> Reaction | None:
+    ) -> list[Reaction]:
         q = f"SELECT {cls.columns} FROM reaction WHERE msg_mxid=$1 AND mx_room=$2 AND tg_sender=$3"
-        return cls._from_row(await cls.db.fetchrow(q, mxid, mx_room, tg_sender))
+        rows = await cls.db.fetch(q, mxid, mx_room, tg_sender)
+        return [cls._from_row(row) for row in rows]
 
     @classmethod
     async def get_all_by_message(cls, mxid: EventID, mx_room: RoomID) -> list[Reaction]:
@@ -68,6 +70,13 @@ class Reaction:
         rows = await cls.db.fetch(q, mxid, mx_room)
         return [cls._from_row(row) for row in rows]
 
+    @property
+    def telegram(self) -> TypeReaction:
+        if self.reaction.isdecimal():
+            return ReactionCustomEmoji(document_id=int(self.reaction))
+        else:
+            return ReactionEmoji(emoticon=self.reaction)
+
     @property
     def _values(self):
         return (
@@ -81,11 +90,11 @@ class Reaction:
     async def save(self) -> None:
         q = """
             INSERT INTO reaction (mxid, mx_room, msg_mxid, tg_sender, reaction)
-            VALUES ($1, $2, $3, $4, $5) ON CONFLICT (msg_mxid, mx_room, tg_sender)
+            VALUES ($1, $2, $3, $4, $5) ON CONFLICT (msg_mxid, mx_room, tg_sender, reaction)
-                DO UPDATE SET mxid=$1, reaction=$5
+                DO UPDATE SET mxid=excluded.mxid
         """
         await self.db.execute(q, *self._values)
 
     async def delete(self) -> None:
-        q = "DELETE FROM reaction WHERE msg_mxid=$1 AND mx_room=$2 AND tg_sender=$3"
+        q = "DELETE FROM reaction WHERE msg_mxid=$1 AND mx_room=$2 AND tg_sender=$3 AND reaction=$4"
-        await self.db.execute(q, self.msg_mxid, self.mx_room, self.tg_sender)
+        await self.db.execute(q, self.msg_mxid, self.mx_room, self.tg_sender, self.reaction)

mautrix_telegram/db/telegram_file.py

@@ -77,11 +77,19 @@ class TelegramFile:
         file = cls._from_row(row)
         if file is None:
             return None
-        thumbnail_id = row.get("thumbnail", None)
+        try:
+            thumbnail_id = row["thumbnail"]
+        except KeyError:
+            thumbnail_id = None
         if thumbnail_id and not _thumbnail:
             file.thumbnail = await cls.get(thumbnail_id, _thumbnail=True)
         return file
 
+    @classmethod
+    async def find_by_mxc(cls, mxc: ContentURI) -> TelegramFile | None:
+        q = f"SELECT {cls.columns} FROM telegram_file WHERE mxc=$1"
+        return cls._from_row(await cls.db.fetchrow(q, mxc))
+
     async def insert(self) -> None:
         q = (
             "INSERT INTO telegram_file (id, mxc, mime_type, was_converted, size, width, height, "

mautrix_telegram/db/upgrade/__init__.py

@@ -15,4 +15,9 @@ from . import (
     v10_more_backfill_fields,
     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

@@ -13,12 +13,12 @@
 #
 # 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 mautrix.util.async_db import Connection, Scheme
 
-latest_version = 10
+latest_version = 17
 
 
-async def create_latest_tables(conn: Connection) -> int:
+async def create_latest_tables(conn: Connection, scheme: Scheme) -> int:
     await conn.execute(
         """CREATE TABLE "user" (
             mxid TEXT   PRIMARY KEY,
@@ -26,6 +26,7 @@ async def create_latest_tables(conn: Connection) -> int:
             tg_username    TEXT,
             tg_phone       TEXT,
             is_bot         BOOLEAN NOT NULL DEFAULT false,
+            is_premium     BOOLEAN NOT NULL DEFAULT false,
             saved_contacts INTEGER NOT NULL DEFAULT 0
         )"""
     )
@@ -66,10 +67,13 @@ async def create_latest_tables(conn: Connection) -> int:
             edit_index   INTEGER,
             redacted     BOOLEAN NOT NULL DEFAULT false,
             content_hash bytea,
+            sender_mxid  TEXT,
+            sender       BIGINT,
             PRIMARY KEY (tgid, tg_space, edit_index),
             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,
@@ -78,7 +82,7 @@ async def create_latest_tables(conn: Connection) -> int:
             tg_sender BIGINT,
             reaction  TEXT NOT NULL,
 
-            PRIMARY KEY (msg_mxid, mx_room, tg_sender),
+            PRIMARY KEY (msg_mxid, mx_room, tg_sender, reaction),
             UNIQUE (mxid, mx_room)
         )"""
     )
@@ -111,6 +115,7 @@ async def create_latest_tables(conn: Connection) -> int:
             avatar_set          BOOLEAN NOT NULL DEFAULT false,
             is_bot              BOOLEAN,
             is_channel          BOOLEAN NOT NULL DEFAULT false,
+            is_premium          BOOLEAN NOT NULL DEFAULT false,
 
             access_token TEXT,
             custom_mxid  TEXT,
@@ -119,6 +124,7 @@ async def create_latest_tables(conn: Connection) -> 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,
@@ -135,6 +141,7 @@ async def create_latest_tables(conn: Connection) -> int:
                 ON UPDATE CASCADE ON DELETE SET NULL
         )"""
     )
+    await conn.execute("CREATE INDEX telegram_file_mxc_idx ON telegram_file(mxc)")
     await conn.execute(
         """CREATE TABLE bot_chat (
             id   BIGINT PRIMARY KEY,
@@ -204,4 +211,31 @@ async def create_latest_tables(conn: Connection) -> int:
             PRIMARY KEY (session_id, entity_id)
         )"""
     )
+    gen = ""
+    if scheme in (Scheme.POSTGRES, Scheme.COCKROACH):
+        gen = "GENERATED ALWAYS AS IDENTITY"
+    await conn.execute(
+        f"""
+        CREATE TABLE backfill_queue (
+            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,
+            dispatch_time       TIMESTAMP,
+            completed_at        TIMESTAMP,
+            cooldown_timeout    TIMESTAMP,
+            FOREIGN KEY (user_mxid) REFERENCES "user"(mxid) ON DELETE CASCADE ON UPDATE CASCADE,
+            FOREIGN KEY (portal_tgid, portal_tg_receiver)
+                REFERENCES portal(tgid, tg_receiver) ON DELETE CASCADE
+        )
+        """
+    )
+
     return latest_version

mautrix_telegram/db/upgrade/v01_initial_revision.py

@@ -24,29 +24,21 @@ legacy_version_query = "SELECT version_num FROM alembic_version"
 last_legacy_version = "bfc0a39bfe02"
 
 
-def table_exists(scheme: str, name: str) -> str:
+async def first_upgrade_target(conn: Connection, scheme: Scheme) -> int:
-    if scheme == Scheme.SQLITE:
+    is_legacy = await conn.table_exists("alembic_version")
-        return f"SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='{name}')"
-    elif scheme in (Scheme.POSTGRES, Scheme.COCKROACH):
-        return f"SELECT EXISTS(SELECT 1 FROM information_schema.tables WHERE table_name='{name}')"
-    raise RuntimeError("unsupported database scheme")
-
-
-async def first_upgrade_target(conn: Connection, scheme: str) -> int:
-    is_legacy = await conn.fetchval(table_exists(scheme, "alembic_version"))
     # If it's a legacy db, the upgrade process will go to v1 and run each migration up to latest.
     # If it's a new db, we'll create the latest tables directly (see create_latest_tables call).
     return 1 if is_legacy else latest_version
 
 
 @upgrade_table.register(description="Initial asyncpg revision", upgrades_to=first_upgrade_target)
-async def upgrade_v1(conn: Connection, scheme: str) -> int:
+async def upgrade_v1(conn: Connection, scheme: Scheme) -> int:
-    is_legacy = await conn.fetchval(table_exists(scheme, "alembic_version"))
+    is_legacy = await conn.table_exists("alembic_version")
     if is_legacy:
         await migrate_legacy_to_v1(conn, scheme)
         return 1
     else:
-        return await create_latest_tables(conn)
+        return await create_latest_tables(conn, scheme)
 
 
 async def drop_constraints(conn: Connection, table: str, contype: str) -> None:
@@ -59,14 +51,14 @@ async def drop_constraints(conn: Connection, table: str, contype: str) -> None:
     await conn.execute(f"ALTER TABLE {table} {drops}")
 
 
-async def migrate_legacy_to_v1(conn: Connection, scheme: str) -> None:
+async def migrate_legacy_to_v1(conn: Connection, scheme: Scheme) -> None:
     legacy_version = await conn.fetchval(legacy_version_query)
     if legacy_version != last_legacy_version:
         raise RuntimeError(
             "Legacy database is not on last version. "
             "Please upgrade the old database with alembic or drop it completely first."
         )
-    if scheme != "sqlite":
+    if scheme != Scheme.SQLITE:
         await drop_constraints(conn, "contact", contype="f")
         await conn.execute(
             """
@@ -131,12 +123,12 @@ async def migrate_legacy_to_v1(conn: Connection, scheme: str) -> None:
     await conn.execute("DROP TABLE alembic_version")
 
 
-async def update_state_store(conn: Connection, scheme: str) -> None:
+async def update_state_store(conn: Connection, scheme: Scheme) -> None:
     # The Matrix state store already has more or less the correct schema, so set the version
     await conn.execute("CREATE TABLE mx_version (version INTEGER PRIMARY KEY)")
     await conn.execute("INSERT INTO mx_version (version) VALUES (2)")
     await conn.execute("UPDATE mx_user_profile SET membership='LEAVE' WHERE membership='LEFT'")
-    if scheme != "sqlite":
+    if scheme != Scheme.SQLITE:
         # Also add the membership type on postgres
         await conn.execute(
             "CREATE TYPE membership AS ENUM ('join', 'leave', 'invite', 'ban', 'knock')"

mautrix_telegram/db/upgrade/v13_multiple_reactions.py

@@ -0,0 +1,54 @@
+# 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="Allow multiple reactions from the same user")
+async def upgrade_v13(conn: Connection, scheme: Scheme) -> None:
+    await conn.execute("CREATE INDEX telegram_file_mxc_idx ON telegram_file(mxc)")
+    await conn.execute('ALTER TABLE "user" ADD COLUMN is_premium BOOLEAN NOT NULL DEFAULT false')
+    await conn.execute("ALTER TABLE puppet ADD COLUMN is_premium BOOLEAN NOT NULL DEFAULT false")
+    if scheme == Scheme.POSTGRES:
+        await conn.execute(
+            """
+            ALTER TABLE reaction
+                DROP CONSTRAINT reaction_pkey,
+                ADD CONSTRAINT reaction_pkey PRIMARY KEY (msg_mxid, mx_room, tg_sender, reaction)
+            """
+        )
+    else:
+        await conn.execute(
+            """CREATE TABLE new_reaction (
+                mxid      TEXT NOT NULL,
+                mx_room   TEXT NOT NULL,
+                msg_mxid  TEXT NOT NULL,
+                tg_sender BIGINT,
+                reaction  TEXT NOT NULL,
+
+                PRIMARY KEY (msg_mxid, mx_room, tg_sender, reaction),
+                UNIQUE (mxid, mx_room)
+            )"""
+        )
+        await conn.execute(
+            """
+            INSERT INTO new_reaction (mxid, mx_room, msg_mxid, tg_sender, reaction)
+            SELECT mxid, mx_room, msg_mxid, tg_sender, reaction FROM reaction
+            """
+        )
+        await conn.execute("DROP TABLE reaction")
+        await conn.execute("ALTER TABLE new_reaction RENAME TO reaction")

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

@@ -37,6 +37,7 @@ class User:
     tg_username: str | None
     tg_phone: str | None
     is_bot: bool
+    is_premium: bool
     saved_contacts: int
 
     @classmethod
@@ -45,7 +46,9 @@ class User:
             return None
         return cls(**row)
 
-    columns: ClassVar[str] = "mxid, tgid, tg_username, tg_phone, is_bot, saved_contacts"
+    columns: ClassVar[str] = ", ".join(
+        ("mxid", "tgid", "tg_username", "tg_phone", "is_bot", "is_premium", "saved_contacts")
+    )
 
     @classmethod
     async def get_by_tgid(cls, tgid: TelegramID) -> User | None:
@@ -78,21 +81,23 @@ class User:
             self.tg_username,
             self.tg_phone,
             self.is_bot,
+            self.is_premium,
             self.saved_contacts,
         )
 
     async def save(self) -> None:
-        q = (
+        q = """
-            'UPDATE "user" SET tgid=$2, tg_username=$3, tg_phone=$4, is_bot=$5, saved_contacts=$6 '
+        UPDATE "user" SET tgid=$2, tg_username=$3, tg_phone=$4, is_bot=$5, is_premium=$6,
-            "WHERE mxid=$1"
+                          saved_contacts=$7
-        )
+        WHERE mxid=$1
+        """
         await self.db.execute(q, *self._values)
 
     async def insert(self) -> None:
-        q = (
+        q = """
-            'INSERT INTO "user" (mxid, tgid, tg_username, tg_phone, is_bot, saved_contacts) '
+        INSERT INTO "user" (mxid, tgid, tg_username, tg_phone, is_bot, is_premium, saved_contacts)
-            "VALUES ($1, $2, $3, $4, $5, $6)"
+        VALUES ($1, $2, $3, $4, $5, $6, $7)
-        )
+        """
         await self.db.execute(q, *self._values)
 
     async def get_contacts(self) -> list[TelegramID]:

mautrix_telegram/example-config.yaml

@@ -7,7 +7,9 @@ homeserver:
     # Whether or not to verify the SSL certificate of the homeserver.
     # Only applies if address starts with https://
     verify_ssl: true
-    asmux: false
+    # What software is the homeserver running?
+    # Standard Matrix homeservers like Synapse, Dendrite and Conduit should just use "standard" here.
+    software: standard
     # Number of retries for all HTTP requests if the homeserver isn't reachable.
     http_retry_count: 4
     # The URL to push real-time bridge status to.
@@ -45,6 +47,7 @@ appservice:
     # https://magicstack.github.io/asyncpg/current/api/index.html#asyncpg.pool.create_pool
     # https://docs.python.org/3/library/sqlite3.html#sqlite3.connect
     # For sqlite, min_size is used as the connection thread pool size and max_size is ignored.
+    # Additionally, SQLite supports init_commands as an array of SQL queries to run on connect (e.g. to set PRAGMAs).
     database_opts:
         min_size: 1
         max_size: 10
@@ -166,7 +169,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.
@@ -221,6 +227,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.
@@ -255,6 +264,8 @@ bridge:
         # Default to encryption, force-enable encryption in all portals the bridge creates
         # This will cause the bridge bot to be in private chats for the encryption to work properly.
         default: false
+        # Whether to use MSC2409/MSC3202 instead of /sync long polling for receiving encryption-related data.
+        appservice: false
         # Require encryption, drop any unencrypted messages.
         require: false
         # Enable key sharing? If enabled, key requests for rooms where users are in will be fulfilled.
@@ -329,34 +340,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
+        # Allow backfilling at all?
-        # invited to private chats when backfilling history from Telegram. This is
+        enable: true
-        # usually needed to prevent rate limits and to allow timestamp massaging.
+        # Use MSC2716 for backfilling?
-        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.
         #
-        # N.B. Initial backfill will only start after member sync. Make sure your
+        # This requires a server with MSC2716 support, which is currently an experimental feature in Synapse.
-        #      max_initial_member_sync is set to a low enough value so it doesn't take forever.
+        # It can be enabled by setting experimental_features -> msc2716_enabled to true in homeserver.yaml.
-        initial_limit: 0
+        msc2716: false
-        # Maximum number of messages to backfill if messages were missed while the bridge was
+        # Use double puppets for backfilling?
-        # disconnected. Note that this only works for logged in users and only if the chat isn't
+        #
-        # older than sync_update_limit
+        # If using MSC2716, the double puppets must be in the appservice's user ID namespace
-        # Set to 0 to disable backfilling missed messages.
+        # (because the bridge can't use the double puppet access token with batch sending).
-        missed_limit: 50
+        #
-        # If using double puppeting, should notifications be disabled
+        # Even without MSC2716, bridging old messages with correct timestamps requires the double
-        # while the initial backfill is in progress?
+        # puppets to be in an appservice namespace, or the server to be modified to allow
-        disable_notifications: false
+        # 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 _get_fwd_entity(client, evt)
-                user = await source.client.get_entity(fwd_from.from_id)
+            if user:
-                if user:
+                fwd_from_text, _ = pu.Puppet.get_displayname(user, False)
-                    fwd_from_text, _ = pu.Puppet.get_displayname(user, False)
+                fwd_from_html = f"<b>{escape(fwd_from_text)}</b>"
-                    fwd_from_html = f"<b>{escape(fwd_from_text)}</b>"
+            else:
-            except (ValueError, RPCError):
                 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 _get_fwd_entity(client, evt)
-                channel = await source.client.get_entity(fwd_from.from_id)
+            if channel:
-                if channel:
+                fwd_from_text = f"channel {channel.title}"
-                    fwd_from_text = f"channel {channel.title}"
+                fwd_from_html = f"channel <b>{escape(channel.title)}</b>"
-                    fwd_from_html = f"channel <b>{escape(channel.title)}</b>"
+            else:
-            except (ValueError, RPCError):
                 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(
+            if isinstance(entity.file, UnicodeCustomEmoji):
-                f'<img data-mx-emoticon data-mau-animated-emoji src="{escape(entity.file.mxc)}" '
+                html.append(entity.file.emoji)
-                f'height="32" width="32" alt="{entity_text}" title="{entity_text}"/>'
+            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,
@@ -388,7 +420,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 +430,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 +457,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,
@@ -486,7 +524,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 +567,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 +633,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 +645,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 +653,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 +681,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))
+                entity = await client.get_entity(PeerUser(contact.user_id))
-                await puppet.update_info(source, entity)
+                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/participants.py

@@ -24,6 +24,7 @@ from telethon.tl.types import (
     ChannelParticipantBanned,
     ChannelParticipantsRecent,
     ChannelParticipantsSearch,
+    ChatParticipantsForbidden,
     InputChannel,
     InputUser,
     TypeChannelParticipant,
@@ -93,6 +94,8 @@ async def get_users(
 ) -> list[TypeUser]:
     if peer_type == "chat":
         chat = await client(GetFullChatRequest(chat_id=tgid))
+        if isinstance(chat.full_chat.participants, ChatParticipantsForbidden):
+            return []
         users = list(_filter_participants(chat.users, chat.full_chat.participants.participants))
         return users[:limit] if limit > 0 else users
     elif peer_type == "channel":

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)
+    if is_initial:
-    for user, user_level in levels.users.items():
+        levels.users.update(userlevel_overrides)
-        if user_level < bot_level:
-            levels.users[user] = userlevel_overrides.get(user, 0)
     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:
@@ -80,6 +80,7 @@ class Puppet(DBPuppet, BasePuppet):
         avatar_set: bool = False,
         is_bot: bool = False,
         is_channel: bool = False,
+        is_premium: bool = False,
         custom_mxid: UserID | None = None,
         access_token: str | None = None,
         next_batch: SyncToken | None = None,
@@ -101,6 +102,7 @@ class Puppet(DBPuppet, BasePuppet):
             avatar_set=avatar_set,
             is_bot=is_bot,
             is_channel=is_channel,
+            is_premium=is_premium,
             custom_mxid=custom_mxid,
             access_token=access_token,
             next_batch=next_batch,
@@ -145,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
@@ -253,13 +252,22 @@ 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)
-        changed = is_bot != self.is_bot or is_channel != self.is_channel
+        changed = (
+            is_bot != self.is_bot or is_channel != self.is_channel or is_premium != self.is_premium
+        )
 
         self.is_bot = is_bot
         self.is_channel = is_channel
+        self.is_premium = is_premium
 
         if self.username != info.username:
             self.username = info.username
@@ -271,8 +279,16 @@ class Puppet(DBPuppet, BasePuppet):
 
         if not self.disable_updates:
             try:
-                changed = await self.update_displayname(source, info) or changed
+                changed = (
-                changed = await self.update_avatar(source, info.photo) or 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}")
 
@@ -287,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
@@ -314,7 +333,7 @@ class Puppet(DBPuppet, BasePuppet):
             return False
 
         if isinstance(info, UpdateUserName):
-            info = await source.client.get_entity(self.peer)
+            info = await (client_override or source.client).get_entity(self.peer)
         if isinstance(info, Channel) or not info.contact:
             self.displayname_contact = False
         elif not self.displayname_contact:
@@ -351,7 +370,11 @@ 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
@@ -370,11 +393,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"],
                 )
@@ -393,7 +427,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,29 @@
 # 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 typing import TYPE_CHECKING, Any, AsyncGenerator, AsyncIterable, Awaitable, NamedTuple, cast
-from datetime import datetime, timezone
+from datetime import datetime
 import asyncio
+import time
 
-from telethon.errors import AuthKeyDuplicatedError, RPCError, UnauthorizedError
+from telethon.errors import (
+    AuthKeyDuplicatedError,
+    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,6 +51,7 @@ 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
@@ -52,9 +61,10 @@ from mautrix.types import PushActionType, PushRuleKind, PushRuleScope, RoomID, R
 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 +93,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,
@@ -92,6 +112,7 @@ class User(DBUser, AbstractUser, BaseUser):
         tg_username: str | None = None,
         tg_phone: str | None = None,
         is_bot: bool = False,
+        is_premium: bool = False,
         saved_contacts: int = 0,
     ) -> None:
         super().__init__(
@@ -100,6 +121,7 @@ class User(DBUser, AbstractUser, BaseUser):
             tg_username=tg_username,
             tg_phone=tg_phone,
             is_bot=is_bot,
+            is_premium=is_premium,
             saved_contacts=saved_contacts,
         )
         AbstractUser.__init__(self)
@@ -109,6 +131,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,
@@ -232,6 +265,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:
@@ -244,6 +285,7 @@ class User(DBUser, AbstractUser, BaseUser):
                     if self._is_backfilling
                     else BridgeStateEvent.CONNECTED,
                     ttl=3600,
+                    info=self._bridge_state_info,
                 )
             else:
                 await self.push_bridge_state(
@@ -268,7 +310,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:
@@ -286,11 +328,16 @@ 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)
 
     async def post_login(self, info: TLUser = None, first_login: bool = False) -> None:
-        if self.config["metrics.enabled"] and not self._track_connection_task:
+        if (
+            self.config["metrics.enabled"] or self.config["homeserver.status_endpoint"]
+        ) and not self._track_connection_task:
             self._track_connection_task = asyncio.create_task(self._track_connection())
 
         try:
@@ -300,6 +347,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)
@@ -309,7 +358,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()
@@ -319,6 +368,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
@@ -369,6 +535,9 @@ class User(DBUser, AbstractUser, BaseUser):
         if self.is_bot != info.bot:
             self.is_bot = info.bot
             changed = True
+        if self.is_premium != info.premium:
+            self.is_premium = info.premium
+            changed = True
         if self.tg_username != info.username:
             self.tg_username = info.username
             changed = True
@@ -414,10 +583,11 @@ class User(DBUser, AbstractUser, BaseUser):
                 pass
             self.tgid = None
         ok = await self.client.log_out()
-        await self.client.session.delete()
+        sess = self.client.session
+        await self.stop()
+        await sess.delete()
         await self.delete()
         self.by_mxid.pop(self.mxid, None)
-        await self.stop()
         self._track_metric(METRIC_LOGGED_IN, False)
         return ok
 
@@ -475,19 +645,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 > time.time():
-        if mute_until is not None and mute_until > now:
             self.log.debug(
                 f"Muting {portal.mxid}/{portal.tgid} (muted until {mute_until} on Telegram)"
             )
@@ -543,15 +712,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:
@@ -559,31 +741,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
+            await self._post_sync_dialog(
-            if dialog.unread_count == 0:
+                portal=portal,
-                # This is usually more reliable than finding a specific message
+                puppet=puppet,
-                # e.g. if the last read message is a service message that isn't in the message db
+                was_created=was_created,
-                last_read = await DBMessage.find_last(portal.mxid, tg_space)
+                **post_sync_args,
-            else:
+            )
-                last_read = await DBMessage.get_one_by_tgid(
+        self.log.debug(f"_sync_dialog finished for {portal.tgid_log}")
-                    portal.tgid, tg_space, dialog.dialog.read_inbox_max_id
+
+    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)"
                 )
+        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, dialog.dialog.notify_settings.mute_until)
+                await self._mute_room(puppet, portal, mute_until)
-                await self._tag_room(
+                await self._tag_room(puppet, portal, self.config["bridge.pinned_tag"], pinned)
-                    puppet, portal, self.config["bridge.pinned_tag"], dialog.pinned
+                await self._tag_room(puppet, portal, self.config["bridge.archive_tag"], archived)
-                )
+        except Exception:
-                await self._tag_room(
+            self.log.exception(f"Error updating read status and tags for {portal.tgid_log}")
-                    puppet, portal, self.config["bridge.archive_tag"], dialog.archived
-                )
 
     async def get_cached_portals(self) -> dict[tuple[TelegramID, TelegramID], po.Portal]:
         if self._portals_cache is None:
@@ -600,9 +824,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(
+        self.log.debug(f"Syncing dialogs ({update_limit=}, {create_limit=})")
-            f"Syncing dialogs (update_limit={update_limit}, create_limit={create_limit})"
-        )
         await self.push_bridge_state(BridgeStateEvent.BACKFILLING)
         puppet = await pu.Puppet.get_by_custom_mxid(self.mxid)
         dialog: Dialog
@@ -623,11 +845,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
@@ -686,8 +909,51 @@ class User(DBUser, AbstractUser, BaseUser):
             await puppet.update_info(self, user)
             contacts[user.id] = puppet.contact_info
         await self.set_contacts(contacts.keys())
+        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_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,11 +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 pickle
+import pkgutil
 import tempfile
 import time
 
@@ -42,9 +44,9 @@ from telethon.tl.types import (
     PhotoSize,
     TypePhotoSize,
 )
-import magic
 
 from mautrix.appservice import IntentAPI
+from mautrix.util import magic, variation_selector
 
 from .. import abstract_user as au
 from ..db import TelegramFile as DBTelegramFile
@@ -177,7 +179,7 @@ async def transfer_thumbnail_to_matrix(
     else:
         file = await client.download_file(thumbnail_loc)
         width, height = None, None
-        mime_type = magic.from_buffer(file, mime=True)
+        mime_type = magic.mimetype(file)
 
     decryption_info = None
     upload_mime_type = mime_type
@@ -212,20 +214,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]
+    source: au.AbstractUser, emoji_ids: list[int], client: MautrixTelegramClient | None = None
-) -> dict[int, DBTelegramFile]:
+) -> 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 +263,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,
@@ -335,13 +361,10 @@ async def _unlocked_transfer_file_to_matrix(
             return None
 
         width, height = None, None
-        mime_type = magic.from_buffer(file, mime=True)
+        mime_type = magic.mimetype(file)
 
         image_converted = False
-        # A weird bug in alpine/magic makes it return application/octet-stream for gzips...
+        is_tgs = mime_type == "application/gzip"
-        is_tgs = mime_type == "application/gzip" or (
-            mime_type == "application/octet-stream" and magic.from_buffer(file).startswith("gzip")
-        )
         if is_sticker and tgs_convert and is_tgs:
             converted_anim = await convert_tgs_to(
                 file, tgs_convert["target"], **tgs_convert["args"]

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

@@ -128,7 +128,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:

mautrix_telegram/web/provisioning/__init__.py

@@ -25,6 +25,7 @@ from telethon.tl.types import ChannelForbidden, ChatForbidden, TypeChat, User as
 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
 
@@ -53,7 +54,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 +63,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,6 +72,8 @@ class ProvisioningAPI(AuthAPI):
         )
         self.app.router.add_route("POST", f"{user_prefix}/pm/{{identifier}}", self.start_dm)
 
+        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("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)
@@ -494,6 +497,22 @@ class ProvisioningAPI(AuthAPI):
             status=201 if just_created else 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 send_bot_token(self, request: web.Request) -> web.Response:
         data, user, err = await self.get_user_request_info(request)
         if err is not None:
@@ -639,6 +658,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:

optional-requirements.txt

@@ -3,7 +3,6 @@
 
 #/speedups
 cryptg>=0.1,<0.4
-cchardet
 aiodns
 brotli
 
@@ -18,7 +17,7 @@ moviepy>=1,<2
 phonenumbers>=8,<9
 
 #/metrics
-prometheus_client>=0.6,<0.15
+prometheus_client>=0.6,<0.16
 
 #/e2be
 python-olm>=3,<4

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.17.8,<0.18
+mautrix>=0.18.8,<0.19
-#telethon>=1.24,<1.25
+#telethon>=1.25.4,<1.27
-tulir-telethon==1.25.0a20
+tulir-telethon==1.27.0a1
-asyncpg>=0.20,<0.27
+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"]),