Bump version to 0.15.2

Add periodic connection refresh

.github/workflows/python-lint.yml

@@ -6,17 +6,17 @@ jobs:
   lint:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v2
-    - uses: actions/setup-python@v2
+    - uses: actions/checkout@v4
+    - uses: actions/setup-python@v5
       with:
-        python-version: "3.11"
+        python-version: "3.12"
     - uses: isort/isort-action@master
       with:
         sortPaths: "./mautrix_telegram"
     - uses: psf/black@stable
       with:
         src: "./mautrix_telegram"
-        version: "23.1.0"
+        version: "24.1.1"
     - name: pre-commit
       run: |
         pip install pre-commit

.gitignore

@@ -10,13 +10,19 @@ __pycache__
 /*.egg-info
 /.eggs
 
-/config.yaml
-/registration.yaml
+*.yaml
+!.pre-commit-config.yaml
+!example-config.yaml
+!/mautrix_telegram/web/provisioning/spec.yaml
+!/.github/workflows/*.yaml
+
+/start
+/mautrix
+/telethon
+
 *.log*
 *.db
 *.db-*
 /*.pickle
 *.bak
-/*.session
-/*.session-journal
 /*.json

.pre-commit-config.yaml

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

CHANGELOG.md

@@ -1,3 +1,43 @@
+# v0.15.2 (2024-07-16)
+
+* Dropped support for Python 3.9.
+* Updated Telegram API to layer 183.
+* Added support for authenticated media downloads.
+* Added support for receiving reactions when using a bot account.
+* Added option to limit file size by chat type.
+* Fixed reply bridging breaking in some cases.
+
+# v0.15.1 (2023-12-26)
+
+* Updated Telegram API to layer 169.
+* Updated Docker image to Alpine 3.19.
+* Fixed some potential cases where a portal room would be created for the
+  relaybot even if `ignore_unbridged_group_chat` was enabled.
+* Fixed member sync in groups with hidden members causing puppeted Matrix users
+  to be kicked even if they're still in the group.
+
+# v0.15.0 (2023-11-26)
+
+* Removed support for MSC2716 backfilling.
+* Added `add-contact` and `delete-contact` commands.
+* Updated Telegram API layer to 166.
+  * Includes receiving view-once media, blockquotes, quote replies and other
+    such things
+* Fixed AuthKeyNotFound errors not being handled and causing users to get stuck
+  in a non-logged-in state.
+
+# v0.14.2 (2023-09-19)
+
+* **Security:** Updated Pillow to 10.0.1.
+* Added support for double puppeting with arbitrary `as_token`s.
+  See [docs](https://docs.mau.fi/bridges/general/double-puppeting.html#appservice-method-new) for more info.
+* Added support for sending webm and tgs files as stickers.
+* Updated to Telegram API layer 161.
+* Fixed cached usernames for Telegram users being cleared incorrectly, leading
+  to mentions not being bridged as usernames.
+* Fixed reaction bridging failing if the server running the bridge was rebooted
+  less than 12 hours ago.
+
 # v0.14.1 (2023-06-26)
 
 ### Added

Dockerfile

@@ -1,9 +1,11 @@
-FROM dock.mau.dev/tulir/lottieconverter:alpine-3.18
+FROM dock.mau.dev/tulir/lottieconverter:alpine-3.19
 
 RUN apk add --no-cache \
       python3 py3-pip py3-setuptools py3-wheel \
       py3-pillow \
       py3-aiohttp \
+      py3-asyncpg \
+      py3-aiosqlite \
       py3-magic \
       py3-ruamel.yaml \
       py3-commonmark \
@@ -15,6 +17,8 @@ RUN apk add --no-cache \
       py3-rsa \
       #py3-telethon \ (outdated)
         py3-pyaes \
+        py3-aiodns \
+        py3-python-socks \
         # cryptg
           py3-cffi \
           py3-qrcode \
@@ -38,13 +42,13 @@ COPY requirements.txt /opt/mautrix-telegram/requirements.txt
 COPY optional-requirements.txt /opt/mautrix-telegram/optional-requirements.txt
 WORKDIR /opt/mautrix-telegram
 RUN apk add --virtual .build-deps python3-dev libffi-dev build-base \
- && pip3 install /cryptg-*.whl \
- && pip3 install --no-cache-dir -r requirements.txt -r optional-requirements.txt \
+ && pip3 install --break-system-packages /cryptg-*.whl \
+ && pip3 install --break-system-packages --no-cache-dir -r requirements.txt -r optional-requirements.txt \
  && apk del .build-deps \
  && rm -f /cryptg-*.whl
 
 COPY . /opt/mautrix-telegram
-RUN apk add git && pip3 install --no-cache-dir .[all] && apk del git \
+RUN apk add git && pip3 install --break-system-packages --no-cache-dir .[all] && apk del git \
   # This doesn't make the image smaller, but it's needed so that the `version` command works properly
   && cp mautrix_telegram/example-config.yaml . && rm -rf mautrix_telegram .git build
 

dev-requirements.txt

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

mautrix_telegram/__init__.py

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

mautrix_telegram/__main__.py

@@ -104,8 +104,6 @@ class TelegramBridge(Bridge):
         self.log.info("Finished re-sending bridge info state events")
 
     def prepare_stop(self) -> None:
-        for puppet in Puppet.by_custom_mxid.values():
-            puppet.stop()
         self.add_shutdown_actions(user.stop() for user in User.by_tgid.values())
         if self.bot:
             self.add_shutdown_actions(self.bot.stop())

mautrix_telegram/abstract_user.py

@@ -40,6 +40,7 @@ from telethon.tl.types import (
     PeerUser,
     PhoneCallRequested,
     TypeUpdate,
+    UpdateBotMessageReaction,
     UpdateChannel,
     UpdateChannelUserTyping,
     UpdateChatDefaultBannedRights,
@@ -240,6 +241,24 @@ class AbstractUser(ABC):
             use_ipv6=self.config["telegram.connection.use_ipv6"],
         )
         self.client.add_event_handler(self._update_catch)
+        self._schedule_reconnect()
+
+    def _schedule_reconnect(self) -> None:
+        reconnect_interval = self.config["telegram.force_refresh_interval_seconds"]
+        if not reconnect_interval or reconnect_interval == 0:
+            return
+        refresh_time = time.time() + reconnect_interval
+        self.log.info(
+            "Scheduling forced reconnect in %d seconds. Connection will be refreshed at %s",
+            reconnect_interval,
+            time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(refresh_time)),
+        )
+        self.loop.call_later(reconnect_interval, lambda: background_task.create(self._reconnect()))
+
+    async def _reconnect(self) -> None:
+        self.log.info("Reconnecting to Telegram...")
+        await self.stop()
+        await self.start()
 
     @abstractmethod
     async def on_signed_out(self, err: UnauthorizedError | AuthKeyError) -> None:
@@ -306,7 +325,18 @@ class AbstractUser(ABC):
     async def start(self, delete_unless_authenticated: bool = False) -> AbstractUser:
         if not self.client:
             await self._init_client()
-        await self.client.connect()
+        attempts = 1
+        while True:
+            try:
+                await self.client.connect()
+            except Exception:
+                attempts += 1
+                if attempts > 10:
+                    raise
+                self.log.exception("Exception connecting to Telegram, retrying in 5s...")
+                await asyncio.sleep(5)
+            else:
+                break
         self.log.debug(f"{'Bot' if self.is_relaybot else self.mxid} connected: {self.connected}")
         return self
 
@@ -352,6 +382,8 @@ class AbstractUser(ABC):
             await self.update_phone_call(update)
         elif isinstance(update, UpdateMessageReactions):
             await self.update_reactions(update)
+        elif isinstance(update, UpdateBotMessageReaction):
+            await self.update_bot_reactions(update)
         elif isinstance(update, (UpdateChatUserTyping, UpdateChannelUserTyping, UpdateUserTyping)):
             await self.update_typing(update)
         elif isinstance(update, UpdateUserStatus):
@@ -456,6 +488,7 @@ class AbstractUser(ABC):
             return
 
         if not portal or not portal.mxid:
+            # TODO This explodes on channels because the field is channel_id
             self.log.debug(f"Dropping own read receipt in unknown chat ({update.peer})")
             return
 
@@ -624,6 +657,12 @@ class AbstractUser(ABC):
             return
         await portal.handle_telegram_reactions(self, TelegramID(update.msg_id), update.reactions)
 
+    async def update_bot_reactions(self, update: UpdateBotMessageReaction) -> None:
+        portal = await po.Portal.get_by_entity(update.peer, tg_receiver=self.tgid)
+        if not portal or not portal.mxid or not portal.allow_bridging:
+            return
+        await portal.handle_telegram_bot_reactions(self, update)
+
     async def update_phone_call(self, update: UpdatePhoneCall) -> None:
         self.log.debug("Phone call update %s", update)
         if not isinstance(update.phone_call, PhoneCallRequested):
@@ -646,7 +685,11 @@ class AbstractUser(ABC):
             await portal.delete_telegram_user(self.tgid, sender=None)
         elif chan := getattr(update, "mau_channel", None):
             if not portal.mxid:
-                background_task.create(self._delayed_create_channel(chan))
+                if (
+                    not self.is_relaybot
+                    or not self.config["bridge.relaybot.ignore_unbridged_group_chat"]
+                ):
+                    background_task.create(self._delayed_create_channel(chan))
             else:
                 self.log.debug("Updating channel info with data fetched by Telethon")
                 await portal.update_info(self, chan)

mautrix_telegram/commands/handler.py

@@ -159,6 +159,7 @@ def command_handler(
     needs_admin: bool = False,
     management_only: bool = False,
     name: str | None = None,
+    aliases: list[str] | None = None,
     help_text: str = "",
     help_args: str = "",
     help_section: HelpSection = None,
@@ -167,6 +168,7 @@ def command_handler(
         _func,
         _handler_class=CommandHandler,
         name=name,
+        aliases=aliases,
         help_text=help_text,
         help_args=help_args,
         help_section=help_section,

mautrix_telegram/commands/portal/admin.py

@@ -39,8 +39,6 @@ async def clear_db_cache(evt: CommandEvent) -> EventID:
         await evt.reply("Cleared portal cache")
     elif section == "puppet":
         pu.Puppet.by_tgid = {}
-        for puppet in pu.Puppet.by_custom_mxid.values():
-            puppet.stop()
         pu.Puppet.by_custom_mxid = {}
         await asyncio.gather(
             *[puppet.try_start() async for puppet in pu.Puppet.all_with_custom_mxid()]
@@ -69,8 +67,6 @@ async def reload_user(evt: CommandEvent) -> EventID:
     if not user:
         return await evt.reply("User not found")
     puppet = await pu.Puppet.get_by_custom_mxid(mxid)
-    if puppet:
-        puppet.stop()
     await user.stop()
     del u.User.by_tgid[user.tgid]
     del u.User.by_mxid[user.mxid]

mautrix_telegram/commands/telegram/misc.py

@@ -18,8 +18,8 @@ from __future__ import annotations
 from typing import cast
 import base64
 import codecs
-import math
 import re
+import shlex
 
 from aiohttp import ClientSession, InvalidURL
 from telethon.errors import (
@@ -29,10 +29,10 @@ from telethon.errors import (
     InviteHashInvalidError,
     InviteRequestSentError,
     OptionsTooMuchError,
-    TakeoutInitDelayError,
     UserAlreadyParticipantError,
 )
 from telethon.tl.functions.channels import JoinChannelRequest
+from telethon.tl.functions.contacts import DeleteByPhonesRequest, ImportContactsRequest
 from telethon.tl.functions.messages import (
     CheckChatInviteRequest,
     GetBotCallbackAnswerRequest,
@@ -42,12 +42,14 @@ from telethon.tl.functions.messages import (
 from telethon.tl.patched import Message
 from telethon.tl.types import (
     InputMediaDice,
+    InputPhoneContact,
     MessageMediaGame,
     MessageMediaPoll,
     TypeInputPeer,
     TypeUpdates,
     User as TLUser,
 )
+from telethon.tl.types.contacts import ImportedContacts
 from telethon.tl.types.messages import BotCallbackAnswer
 
 from mautrix.types import EventID, Format
@@ -161,6 +163,76 @@ async def pm(evt: CommandEvent) -> EventID:
     return await evt.reply(f"Created private chat room with {displayname}")
 
 
+async def _handle_contact(source: AbstractUser, user: TLUser) -> str:
+    puppet: pu.Puppet = await pu.Puppet.get_by_tgid(user.id)
+    await puppet.update_info(source, user)
+
+    params = []
+    if user.username:
+        params.append(f"[@{user.username}](https://t.me/{user.username})")
+    if user.phone:
+        params.append(f"+{user.phone}")
+    params.append(f"ID `{user.id}`")
+    params_str = " / ".join(params)
+    return f"[{puppet.displayname}](https://matrix.to/#/{puppet.mxid}): {params_str}"
+
+
+@command_handler(
+    help_section=SECTION_CREATING_PORTALS,
+    help_args="<_phone_> <_first name_> <_last name_>",
+    help_text="Add a phone number to your contacts on Telegram",
+)
+async def add_contact(evt: CommandEvent) -> EventID:
+    if len(evt.args) < 3:
+        return await evt.reply(
+            "**Usage:** `$cmdprefix+sp add-contact <phone> <first name> <last name>`"
+        )
+    try:
+        names = shlex.split(" ".join(evt.args[1:]))
+    except ValueError as e:
+        return await evt.reply(
+            f"Failed to parse names (use shell quoting for names with spaces): {e}"
+        )
+    if len(names) != 2:
+        return await evt.reply(
+            "Wrong number of names, must have first and last name "
+            "(use shell quoting for names with spaces)"
+        )
+    res: ImportedContacts = await evt.sender.client(
+        ImportContactsRequest(
+            contacts=[
+                InputPhoneContact(
+                    client_id=1, phone=evt.args[0], first_name=names[0], last_name=names[1]
+                )
+            ]
+        )
+    )
+    if res.retry_contacts:
+        return await evt.reply("Failed to import contacts")
+    elif not res.users:
+        return await evt.reply("Contact imported, but user not found on Telegram")
+    imported_str = "\n".join(
+        [f"* {await _handle_contact(evt.sender, user)}" for user in res.users]
+    )
+    return await evt.reply(f"Imported contacts:\n\n{imported_str}")
+
+
+@command_handler(
+    help_section=SECTION_CREATING_PORTALS,
+    help_args="<_phones..._>",
+    help_text="Remove phone numbers from your contacts on Telegram.",
+    aliases=["remove-contact", "delete-contacts", "remove-contacts"],
+)
+async def delete_contact(evt: CommandEvent) -> EventID:
+    if len(evt.args) == 0:
+        return await evt.reply("**Usage:** `$cmdprefix+sp delete-contact <phones...>`")
+    ok = await evt.sender.client(DeleteByPhonesRequest(phones=evt.args))
+    if ok:
+        return await evt.reply("Contacts deleted")
+    else:
+        return await evt.reply("Contacts not deleted?")
+
+
 async def _join(
     evt: CommandEvent, identifier: str, link_type: str
 ) -> tuple[TypeUpdates | None, EventID | None]:
@@ -440,14 +512,5 @@ 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
-    if portal.backfill_msc2716:
-        messages_per_batch = evt.config["bridge.backfill.incremental.messages_per_batch"]
-        batches = math.ceil(limit / messages_per_batch)
-        rounded = ""
-        if batches * messages_per_batch != limit:
-            rounded = f" (rounded message limit to {batches}*{messages_per_batch})"
-        await portal.enqueue_backfill(evt.sender, priority=0, max_batches=batches)
-        await evt.reply(f"Backfill queued{rounded}")
-    else:
-        output = await portal.forward_backfill(evt.sender, initial=False, override_limit=limit)
-        await evt.reply(output)
+    output = await portal.forward_backfill(evt.sender, initial=False, override_limit=limit)
+    await evt.reply(output)

mautrix_telegram/config.py

@@ -136,6 +136,8 @@ class Config(BaseBridgeConfig):
         copy("bridge.caption_in_message")
         copy("bridge.image_as_file_size")
         copy("bridge.image_as_file_pixels")
+        copy("bridge.document_as_link_size.bot")
+        copy("bridge.document_as_link_size.channel")
         copy("bridge.parallel_file_transfer")
         copy("bridge.federate_rooms")
         copy("bridge.always_custom_emoji_reaction")
@@ -157,6 +159,7 @@ class Config(BaseBridgeConfig):
             if base["bridge.private_chat_portal_meta"] not in ("default", "always", "never"):
                 base["bridge.private_chat_portal_meta"] = "default"
         copy("bridge.disable_reply_fallbacks")
+        copy("bridge.cross_room_replies")
         copy("bridge.delivery_receipts")
         copy("bridge.delivery_error_reports")
         copy("bridge.incoming_bridge_error_reports")
@@ -170,8 +173,6 @@ class Config(BaseBridgeConfig):
         copy("bridge.kick_on_logout")
         copy("bridge.always_read_joined_telegram_notice")
         copy("bridge.backfill.enable")
-        copy("bridge.backfill.msc2716")
-        copy("bridge.backfill.double_puppet_backfill")
         copy("bridge.backfill.normal_groups")
         copy("bridge.backfill.unread_hours_threshold")
         if "bridge.backfill.forward" in self:
@@ -194,6 +195,7 @@ class Config(BaseBridgeConfig):
             copy("bridge.backfill.forward_limits.sync.normal_group")
             copy("bridge.backfill.forward_limits.sync.supergroup")
             copy("bridge.backfill.forward_limits.sync.channel")
+        copy("bridge.backfill.forward_timeout")
         copy("bridge.backfill.incremental.messages_per_batch")
         copy("bridge.backfill.incremental.post_batch_delay")
         copy("bridge.backfill.incremental.max_batches.user")
@@ -262,6 +264,7 @@ class Config(BaseBridgeConfig):
         copy("telegram.catch_up")
         copy("telegram.sequential_updates")
         copy("telegram.exit_on_update_error")
+        copy("telegram.force_refresh_interval_seconds")
 
         copy("telegram.connection.timeout")
         copy("telegram.connection.retries")

mautrix_telegram/db/telethon_session.py

@@ -208,9 +208,9 @@ class PgSession(MemorySession):
             await self._locked_process_entities(tlo)
 
     async def _locked_process_entities(self, tlo) -> None:
-        rows: list[
-            tuple[str, int, int, str | None, str | None, str | None]
-        ] = self._entities_to_rows(tlo)
+        rows: list[tuple[str, int, int, str | None, str | None, str | None]] = (
+            self._entities_to_rows(tlo)
+        )
         if not rows:
             return
         if self.db.scheme == Scheme.POSTGRES:

mautrix_telegram/example-config.yaml

@@ -222,6 +222,11 @@ bridge:
     image_as_file_size: 10
     # Maximum number of pixels in an image before sending to Telegram as a document. Defaults to 4096x4096 = 16777216.
     image_as_file_pixels: 16777216
+    # Maximum size of Telegram documents before linking to Telegrm instead of bridge
+    # to Matrix media.
+    document_as_link_size:
+        channel:
+        bot:
     # Enable experimental parallel file transfer, which makes uploads/downloads much faster by
     # streaming from/to Matrix and using many connections for Telegram.
     # Note that generating HQ thumbnails for videos is not possible with streamed transfers.
@@ -342,6 +347,8 @@ bridge:
     # Disable generating reply fallbacks? Some extremely bad clients still rely on them,
     # but they're being phased out and will be completely removed in the future.
     disable_reply_fallbacks: false
+    # Should cross-chat replies from Telegram be bridged? Most servers and clients don't support this.
+    cross_room_replies: false
     # Whether or not the bridge should send a read receipt from the bridge bot when a message has
     # been sent to Telegram.
     delivery_receipts: false
@@ -377,23 +384,6 @@ bridge:
     backfill:
         # Allow backfilling at all?
         enable: true
-        # Use MSC2716 for backfilling?
-        #
-        # This requires a server with MSC2716 support, which is currently an experimental feature in Synapse.
-        # It can be enabled by setting experimental_features -> msc2716_enabled to true in homeserver.yaml.
-        msc2716: false
-        # Use double puppets for backfilling?
-        #
-        # If using MSC2716, the double puppets must be in the appservice's user ID namespace
-        # (because the bridge can't use the double puppet access token with batch sending).
-        #
-        # Even without MSC2716, bridging old messages with correct timestamps requires the double
-        # puppets to be in an appservice namespace, or the server to be modified to allow
-        # overriding timestamps anyway.
-        #
-        # Also note that adding users to the appservice namespace may have unexpected side effects,
-        # as described in https://docs.mau.fi/bridges/general/double-puppeting.html#appservice-method
-        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.
@@ -403,10 +393,9 @@ bridge:
         # Set to -1 to let any chat be unread.
         unread_hours_threshold: 720
 
-        # Forward backfilling limits. These apply to both MSC2716 and legacy backfill.
+        # Forward backfilling limits.
         #
         # 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_limits:
             # Number of messages to backfill immediately after creating a portal.
             initial:
@@ -420,8 +409,10 @@ bridge:
                 normal_group: 100
                 supergroup: 100
                 channel: 100
+        # Timeout for forward backfills in seconds. If you have a high limit, you'll have to increase this too.
+        forward_timeout: 900
 
-        # Settings for incremental backfill of history. These only apply when using MSC2716.
+        # Settings for incremental backfill of history. These only apply to Beeper, as upstream abandoned MSC2716.
         incremental:
             # Maximum number of messages to backfill per batch.
             messages_per_batch: 100
@@ -590,6 +581,8 @@ telegram:
     # Should incoming updates be handled sequentially to make sure order is preserved on Matrix?
     sequential_updates: true
     exit_on_update_error: false
+    # Interval to force refresh the connection (full reconnect). 0 disables it.
+    force_refresh_interval_seconds: 0
 
     # Telethon connection options.
     connection:

mautrix_telegram/formatter/__init__.py

@@ -1,2 +1,2 @@
 from .from_matrix import matrix_reply_to_telegram, matrix_to_telegram
-from .from_telegram import telegram_to_matrix
+from .from_telegram import telegram_text_to_matrix_html, telegram_to_matrix

mautrix_telegram/formatter/from_matrix/parser.py

@@ -82,14 +82,6 @@ class MatrixParser(BaseMatrixParser[TelegramMessage]):
         prefix = "#" * length + " "
         return TelegramMessage.join(children, "").prepend(prefix).format(TelegramEntityType.BOLD)
 
-    async def blockquote_to_fstring(
-        self, node: HTMLNode, ctx: RecursionContext
-    ) -> TelegramMessage:
-        msg = await self.tag_aware_parse_node(node, ctx)
-        children = msg.trim().split("\n")
-        children = [child.prepend("> ") for child in children]
-        return TelegramMessage.join(children, "\n")
-
     async def color_to_fstring(self, msg: TelegramMessage, color: str) -> TelegramMessage:
         return msg
 

mautrix_telegram/formatter/from_telegram.py

@@ -176,6 +176,21 @@ async def _convert_custom_emoji(
                 entities[i] = ReuploadedCustomEmoji(entity, custom_emojis[entity.document_id])
 
 
+async def telegram_text_to_matrix_html(
+    source: au.AbstractUser,
+    text: str,
+    entities: list[TypeMessageEntity],
+    client: MautrixTelegramClient | None = None,
+) -> str:
+    if not entities:
+        return escape(text).replace("\n", "<br/>")
+    await _convert_custom_emoji(source, entities, client=client)
+    text = add_surrogate(text)
+    html = await _telegram_entities_to_matrix_catch(text, entities)
+    html = del_surrogate(html)
+    return html
+
+
 async def telegram_to_matrix(
     evt: Message | SponsoredMessage,
     source: au.AbstractUser,
@@ -192,10 +207,10 @@ async def telegram_to_matrix(
     )
     entities = override_entities or evt.entities
     if 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)
+        content.formatted_body = await telegram_text_to_matrix_html(
+            source, content.body, entities, client=client
+        )
 
     if require_html:
         content.ensure_has_html()
@@ -333,7 +348,11 @@ async def _telegram_entities_to_matrix(
         last_offset = relative_offset + (0 if skip_entity else entity.length)
     html.append(text_to_html(text[last_offset:]))
 
-    return "".join(html)
+    html_string = "".join(html)
+    # Remove redundant <br>'s after block tags
+    html_string = html_string.replace("</blockquote><br/>", "</blockquote>")
+    html_string = html_string.replace("</pre><br/>", "</pre>")
+    return html_string
 
 
 def _parse_pre(html: list[str], entity_text: str, language: str) -> bool:

mautrix_telegram/matrix.py

@@ -61,21 +61,6 @@ 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,
@@ -174,7 +159,8 @@ class MatrixHandler(BaseMatrixHandler):
             await portal.main_intent.kick_user(
                 room_id,
                 user.mxid,
-                "This chat does not have a bot relaying messages for unauthenticated users.",
+                "This chat does not have a bot on the Telegram side for relaying messages sent by"
+                " unauthenticated Matrix users.",
             )
             return
 

mautrix_telegram/portal.py

@@ -23,6 +23,7 @@ from typing import (
     Callable,
     List,
     Literal,
+    NamedTuple,
     Union,
     cast,
 )
@@ -49,6 +50,7 @@ from telethon.errors import (
     InputUserDeactivatedError,
     MessageEmptyError,
     MessageIdInvalidError,
+    MessageNotModifiedError,
     MessageTooLongError,
     PhotoExtInvalidError,
     PhotoInvalidDimensionsError,
@@ -99,6 +101,7 @@ from telethon.tl.types import (
     DocumentAttributeAudio,
     DocumentAttributeFilename,
     DocumentAttributeImageSize,
+    DocumentAttributeSticker,
     DocumentAttributeVideo,
     GeoPoint,
     InputChannel,
@@ -110,7 +113,9 @@ from telethon.tl.types import (
     InputPeerChat,
     InputPeerPhotoFileLocation,
     InputPeerUser,
+    InputStickerSetEmpty,
     InputUser,
+    MessageActionBoostApply,
     MessageActionChannelCreate,
     MessageActionChatAddUser,
     MessageActionChatCreate,
@@ -157,6 +162,7 @@ from telethon.tl.types import (
     TypeUser,
     TypeUserFull,
     TypeUserProfilePhoto,
+    UpdateBotMessageReaction,
     UpdateChannelUserTyping,
     UpdateChatUserTyping,
     UpdateMessageReactions,
@@ -250,7 +256,6 @@ if TYPE_CHECKING:
 StateBridge = EventType.find("m.bridge", EventType.Class.STATE)
 StateHalfShotBridge = EventType.find("uk.half-shot.bridge", EventType.Class.STATE)
 DummyPortalCreated = EventType.find("fi.mau.dummy.portal_created", EventType.Class.MESSAGE)
-StateMarker = EventType.find("org.matrix.msc2716.marker", EventType.Class.STATE)
 
 InviteList = Union[UserID, List[UserID]]
 UpdateTyping = Union[UpdateUserTyping, UpdateChatUserTyping, UpdateChannelUserTyping]
@@ -268,6 +273,11 @@ class IgnoredMessageError(Exception):
     pass
 
 
+class WrappedReaction(NamedTuple):
+    reaction: ReactionEmoji | ReactionCustomEmoji
+    date: datetime | None
+
+
 class Portal(DBPortal, BasePortal):
     bot: "Bot"
     config: Config
@@ -297,8 +307,6 @@ class Portal(DBPortal, BasePortal):
 
     backfill_lock: SimpleLock
     backfill_method_lock: asyncio.Lock
-    backfill_leave: set[IntentAPI] | None
-    backfill_msc2716: bool
     backfill_enable: bool
 
     alias: RoomAlias | None
@@ -319,7 +327,6 @@ class Portal(DBPortal, BasePortal):
     _sponsored_seen: dict[UserID, bool]
     _new_messages_after_sponsored: bool
 
-    _member_list_cache: dict[EventID, set[UserID]]
     _prev_reaction_poll: dict[UserID, float]
 
     _msg_conv: putil.TelegramMessageConverter
@@ -378,7 +385,6 @@ class Portal(DBPortal, BasePortal):
             "Waiting for backfilling to finish before handling %s", log=self.log
         )
         self.backfill_method_lock = asyncio.Lock()
-        self.backfill_leave = None
 
         self.dedup = putil.PortalDedup(self)
         self.send_lock = putil.PortalSendLock()
@@ -393,7 +399,6 @@ class Portal(DBPortal, BasePortal):
         self._new_messages_after_sponsored = True
         self._bridging_blocked_at_runtime = False
 
-        self._member_list_cache = {}
         self._prev_reaction_poll = defaultdict(lambda: 0.0)
 
         self._msg_conv = putil.TelegramMessageConverter(self)
@@ -439,6 +444,10 @@ class Portal(DBPortal, BasePortal):
     def is_direct(self) -> bool:
         return self.peer_type == "user"
 
+    @property
+    def is_channel(self) -> bool:
+        return self.peer_type == "channel"
+
     @property
     def has_bot(self) -> bool:
         return bool(self.bot) and (
@@ -490,7 +499,6 @@ class Portal(DBPortal, BasePortal):
         cls.filter_list = cls.config["bridge.filter.list"]
         cls.filter_users = cls.config["bridge.filter.users"]
         cls.hs_domain = cls.config["homeserver.domain"]
-        cls.backfill_msc2716 = cls.config["bridge.backfill.msc2716"]
         cls.backfill_enable = cls.config["bridge.backfill.enable"]
         cls.alias_template = SimpleTemplate(
             cls.config["bridge.alias_template"],
@@ -792,6 +800,8 @@ class Portal(DBPortal, BasePortal):
                 background_task.create(update)
                 await self.invite_to_matrix(invites or [])
             return self.mxid
+        elif user.is_relaybot and self.config["bridge.relaybot.ignore_unbridged_group_chat"]:
+            raise Exception("create_matrix_room called as relaybot")
         async with self._room_create_lock:
             try:
                 return await self._create_matrix_room(
@@ -1063,18 +1073,14 @@ class Portal(DBPortal, BasePortal):
             self.first_event_id = await self.main_intent.send_message_event(
                 self.mxid, DummyPortalCreated, {}
             )
-            if not self.bridge.homeserver_software.is_hungry:
-                self._member_list_cache[self.first_event_id] = set(
-                    (await self.main_intent.get_joined_members(self.mxid)).keys()
-                )
             await self.save()
 
-            if self.backfill_enable and (isinstance(user, u.User) or not self.backfill_msc2716):
+            if self.backfill_enable:
                 try:
                     await self.forward_backfill(user, initial=True, client=client)
                 except Exception:
                     self.log.exception("Error in initial backfill")
-                if self.backfill_msc2716:
+                if self._enable_batch_sending:
                     await self.enqueue_backfill(user, priority=50)
 
         return self.mxid
@@ -1153,12 +1159,19 @@ class Portal(DBPortal, BasePortal):
         # We can't trust the member list if any of the following cases is true:
         #  * There are close to 10 000 users, because Telegram might not be sending all members.
         #  * The member sync count is limited, because then we might ignore some members.
-        #  * It's a channel, because non-admins don't have access to the member list.
+        #  * It's a channel, because non-admins don't have access to the member list
+        #    and even admins can only see 200 members.
+        #  * The source user is not in the chat, because that likely means it's a group
+        #    with the member list hidden (so only admins are visible).
         trust_member_list = (
-            len(allowed_tgids) < 9900
-            if self.max_initial_member_sync < 0
-            else len(allowed_tgids) < self.max_initial_member_sync - 10
-        ) and (self.megagroup or self.peer_type != "channel")
+            (
+                len(allowed_tgids) < 9900
+                if self.max_initial_member_sync < 0
+                else len(allowed_tgids) < self.max_initial_member_sync - 10
+            )
+            and (self.megagroup or self.peer_type != "channel")
+            and source.tgid in allowed_tgids
+        )
         if not trust_member_list:
             return None
 
@@ -1841,6 +1854,7 @@ class Portal(DBPortal, BasePortal):
         max_image_size = self.config["bridge.image_as_file_size"] * 1000**2
         max_image_pixels = self.config["bridge.image_as_file_pixels"]
 
+        attributes = []
         if self.config["bridge.parallel_file_transfer"] and content.url:
             file_handle, file_size = await util.parallel_transfer_to_telegram(
                 client, self.main_intent, content.url, sender_id
@@ -1860,21 +1874,27 @@ class Portal(DBPortal, BasePortal):
                 file = await self.main_intent.download_media(content.url)
 
             if content.msgtype == MessageType.STICKER:
-                if mime != "image/gif":
-                    mime, file, w, h = util.convert_image(
-                        file, source_mime=mime, target_type="webp"
-                    )
-                else:
+                if mime == "image/gif":
                     # Remove sticker description
                     file_name = "sticker.gif"
+                else:
+                    if mime not in ("video/webm", "application/x-tgsticker"):
+                        mime, file, w, h = util.convert_image(
+                            file, source_mime=mime, target_type="webp"
+                        )
+                    attributes.append(
+                        DocumentAttributeSticker(
+                            alt=content.body, stickerset=InputStickerSetEmpty()
+                        )
+                    )
 
             file_handle = await client.upload_file(file)
             file_size = len(file)
 
         file_handle.name = file_name
         force_document = file_size >= max_image_size
+        attributes.append(DocumentAttributeFilename(file_name=file_name))
 
-        attributes = [DocumentAttributeFilename(file_name=file_name)]
         if content.msgtype == MessageType.VIDEO:
             attributes.append(
                 DocumentAttributeVideo(
@@ -2047,7 +2067,7 @@ class Portal(DBPortal, BasePortal):
         response: TypeMessage,
         msgtype: MessageType | None = None,
     ) -> None:
-        self.log.trace("Handled Matrix message: %s", response)
+        self.log.trace("Raw event handling response for %s: %s", event_id, response)
         event_hash, _ = self.dedup.check(response, (event_id, space), force_hash=edit_index != 0)
         if edit_index < 0:
             prev_edit = await DBMessage.get_one_by_tgid(TelegramID(response.id), space, -1)
@@ -2077,6 +2097,9 @@ class Portal(DBPortal, BasePortal):
                 seconds=response.ttl_period,
                 expires_at=int(response.date.timestamp()) + response.ttl_period,
             )
+        self.log.debug(
+            f"Handled Matrix message {event_id} -> {response.id} (edit index {edit_index})"
+        )
 
     @staticmethod
     def _error_to_human_message(err: Exception) -> str | None:
@@ -2123,7 +2146,7 @@ class Portal(DBPortal, BasePortal):
             status.status = MessageStatus.FAIL
         elif err:
             status.reason = MessageStatusReason.GENERIC_ERROR
-            status.error = f"{type(err)}: {err}"
+            status.error = f"{type(err).__name__}: {err}"
             status.status = MessageStatus.RETRIABLE
             status.message = self._error_to_human_message(err)
         else:
@@ -2154,9 +2177,10 @@ class Portal(DBPortal, BasePortal):
         )
 
         if msg and self.config["bridge.delivery_error_reports"]:
-            await self._send_message(
-                self.main_intent, TextMessageEventContent(msgtype=MessageType.NOTICE, body=msg)
-            )
+            if not isinstance(err, MessageNotModifiedError):
+                await self._send_message(
+                    self.main_intent, TextMessageEventContent(msgtype=MessageType.NOTICE, body=msg)
+                )
         await self._send_message_status(event_id, err)
 
     async def handle_matrix_message(
@@ -2174,7 +2198,6 @@ class Portal(DBPortal, BasePortal):
                 message_type=content.msgtype,
                 msg=f"\u26a0 Your message may not have been bridged: {e}",
             )
-            raise
         except Exception as e:
             if isinstance(e, IgnoredMessageError):
                 self.log.debug(f"Ignored {event_id}: {e}")
@@ -2313,20 +2336,17 @@ class Portal(DBPortal, BasePortal):
                 sender.command_status = None
             except (KeyError, TypeError):
                 if not logged_in or (
-                    "filename" in content and content["filename"] != content.body
+                    content.filename is not None and content.filename != content.body
                 ):
-                    if "filename" in content:
-                        file_name = content["filename"]
+                    if content.filename:
+                        file_name = content.filename
                     caption_content = TextMessageEventContent(
                         msgtype=MessageType.TEXT,
                         body=content.body,
                     )
-                    if (
-                        "formatted_body" in content
-                        and str(content.get("format")) == Format.HTML.value
-                    ):
-                        caption_content["formatted_body"] = content["formatted_body"]
-                        caption_content["format"] = Format.HTML
+                    if content.formatted_body and content.format == Format.HTML:
+                        caption_content.formatted_body = content.formatted_body
+                        caption_content.format = Format.HTML
                 else:
                     caption_content = None
             if caption_content:
@@ -2420,6 +2440,10 @@ class Portal(DBPortal, BasePortal):
             await deleter.client(
                 SendReactionRequest(peer=self.peer, msg_id=msg.tgid, reaction=new_reactions)
             )
+            self.log.debug(
+                f"Handled Matrix deletion of reaction {event_id} to {msg.tgid} "
+                f"(new reaction count: {len(new_reactions) if new_reactions else 0})"
+            )
 
     async def _handle_matrix_deletion(self, deleter: u.User, event_id: EventID) -> None:
         real_deleter = deleter if not await deleter.needs_relaybot(self) else self.bot
@@ -2440,6 +2464,7 @@ class Portal(DBPortal, BasePortal):
         else:
             await message.mark_redacted()
             await real_deleter.client.delete_messages(self.peer, [message.tgid])
+            self.log.debug(f"Handled Matrix redaction of {event_id} / {message.tgid}")
 
     async def handle_matrix_reaction(
         self, user: u.User, target_event_id: EventID, emoji: str, reaction_event_id: EventID
@@ -2546,9 +2571,15 @@ class Portal(DBPortal, BasePortal):
             SendReactionRequest(peer=self.peer, msg_id=msg.tgid, reaction=new_tg_reactions)
         )
         puppet = await user.get_puppet()
+        removed = 0
         for db_reaction in reactions_to_remove:
+            removed += 1
             await db_reaction.delete()
             await puppet.intent_for(self).redact(db_reaction.mx_room, db_reaction.mxid)
+        self.log.debug(
+            f"Handled Matrix reaction {reaction_event_id} to {msg.tgid} "
+            f"(new reaction count: {len(new_tg_reactions)}, removed {removed} old reactions)"
+        )
         await DBReaction(
             mxid=reaction_event_id,
             mx_room=self.mxid,
@@ -2779,7 +2810,7 @@ class Portal(DBPortal, BasePortal):
         intent = sender.intent_for(self) if sender else self.main_intent
         is_bot = sender.is_bot if sender else False
         converted = await self._msg_conv.convert(
-            source, intent, is_bot, evt, no_reply_fallback=True
+            source, intent, is_bot, self.is_channel, evt, no_reply_fallback=True
         )
         converted.content.set_edit(editing_msg.mxid)
         await intent.set_typing(self.mxid, timeout=0)
@@ -2814,6 +2845,10 @@ class Portal(DBPortal, BasePortal):
     def _default_max_batches(self) -> int:
         return self.config[f"bridge.backfill.incremental.max_batches.{self._backfill_config_type}"]
 
+    @property
+    def _enable_batch_sending(self) -> bool:
+        return self.bridge.matrix.versions.supports("com.beeper.batch_sending")
+
     async def enqueue_backfill(
         self,
         source: u.User,
@@ -2862,13 +2897,15 @@ class Portal(DBPortal, BasePortal):
         )
         if limit == 0:
             return "Limit is zero, not backfilling"
+        timeout = self.config["bridge.backfill.forward_timeout"]
         with self.backfill_lock:
-            output = await asyncio.wait_for(
-                self.backfill(
-                    source, client, forward=True, forward_limit=limit, last_tgid=last_tgid
-                ),
-                timeout=15 * 60,
+            task = self.backfill(
+                source, client, forward=True, forward_limit=limit, last_tgid=last_tgid
             )
+            if timeout > 0:
+                output = await asyncio.wait_for(task, timeout=timeout)
+            else:
+                output = await task
             self.log.debug(f"Forward backfill complete, status: {output}")
             return output
 
@@ -2901,14 +2938,9 @@ class Portal(DBPortal, BasePortal):
         if not self.config["bridge.backfill.normal_groups"] and self.peer_type == "chat":
             return "Backfilling normal groups is disabled in the bridge config"
         tg_space = source.tgid if self.peer_type != "channel" else self.tgid
-        prev_event_id = self.first_event_id
         if forward:
             last_in_room = await DBMessage.find_last(self.mxid, tg_space)
-            if last_in_room:
-                prev_event_id = last_in_room.mxid
-                min_id = last_in_room.tgid
-            else:
-                min_id = 0
+            min_id = last_in_room.tgid if last_in_room else 0
             if last_tgid is None:
                 messages = await source.client.get_messages(self.peer, limit=1)
                 if not messages:
@@ -2939,29 +2971,10 @@ class Portal(DBPortal, BasePortal):
                 f"Backfilling up to {req.messages_per_batch} historical messages "
                 f"before {anchor_id} ({anchor_source}) through {source.mxid}"
             )
-        insertion_id, event_count, message_count, lowest_id = await self._backfill_messages(
-            source, client, forward, anchor_id, limit, prev_event_id
+        event_count, message_count, lowest_id = await self._backfill_messages(
+            source, client, forward, anchor_id, limit
         )
-        if prev_event_id == self.first_event_id:
-            if insertion_id and not self.base_insertion_id:
-                self.base_insertion_id = insertion_id
-            elif not insertion_id:
-                insertion_id = self.base_insertion_id
         await self.save()
-        if (
-            event_count > 0
-            and self.backfill_msc2716
-            and (not forward or not self.bridge.homeserver_software.is_hungry)
-        ):
-            await self.main_intent.send_state_event(
-                self.mxid,
-                StateMarker,
-                {
-                    "org.matrix.msc2716.marker.insertion": insertion_id,
-                    "com.beeper.timestamp": int(time.time() * 1000),
-                },
-                state_key=insertion_id,
-            )
         if forward:
             self.log.debug(f"Forward backfill finished with {event_count}/{message_count} events")
         elif message_count > 0 and lowest_id and lowest_id > 1:
@@ -2980,42 +2993,11 @@ class Portal(DBPortal, BasePortal):
             self.log.debug("No more messages to backfill")
         return f"Backfilled {event_count} messages"
 
-    def _can_double_puppet_backfill(self, custom_mxid: UserID) -> bool:
-        if not self.backfill_msc2716:
-            return True
-        if not self.config["bridge.backfill.double_puppet_backfill"]:
-            return False
-        if self.bridge.homeserver_software.is_hungry:
-            return True
-
-        # Batch sending can only use local users, so don't allow double puppets on other servers.
-        if custom_mxid[custom_mxid.index(":") + 1 :] != self.config["homeserver.domain"]:
-            return False
-        return True
-
-    async def _get_members_at(self, event_id: EventID) -> set[UserID]:
-        try:
-            return self._member_list_cache[event_id]
-        except KeyError:
-            pass
-        # TODO cache the list in db?
-        self.log.debug(f"Fetching member list at {event_id}")
-        ctx = await self.main_intent.get_event_context(self.mxid, event_id, limit=0)
-        members = {
-            evt.state_key
-            for evt in ctx.state
-            if evt.type == EventType.ROOM_MEMBER and evt.content.membership == Membership.JOIN
-        }
-        self.log.debug(f"Found {len(members)} members at {event_id}")
-        self._member_list_cache[event_id] = members
-        return members
-
     async def _convert_batch_msg(
         self,
         source: u.User,
         client: MautrixTelegramClient,
         msg: Message,
-        add_member: Callable[[IntentAPI, str, ContentURI], Awaitable[None]],
     ) -> tuple[putil.ConvertedMessage, IntentAPI]:
         if msg.from_id and isinstance(msg.from_id, (PeerUser, PeerChannel)):
             sender = await p.Puppet.get_by_peer(msg.from_id)
@@ -3033,15 +3015,18 @@ class Portal(DBPortal, BasePortal):
                 await sender.update_info(source, entity, client_override=client)
         else:
             intent = self.main_intent
-        if intent.api.is_real_user and not self._can_double_puppet_backfill(intent.mxid):
+        if (
+            intent.api.is_real_user
+            and not intent.api.is_real_user_as_token
+            and not self._enable_batch_sending
+        ):
             intent = sender.default_mxid_intent
-        if sender:
-            await add_member(intent, sender.displayname, sender.avatar_url)
         is_bot = sender.is_bot if sender else False
         converted = await self._msg_conv.convert(
             source,
             intent,
             is_bot,
+            self.is_channel,
             msg,
             client=client,
             deterministic_reply_id=self.bridge.homeserver_software.is_hungry,
@@ -3081,50 +3066,15 @@ class Portal(DBPortal, BasePortal):
         forward: bool,
         anchor_id: int,
         limit: int,
-        prev_event_id: EventID,
-    ) -> tuple[EventID | None, int, int, TelegramID]:
+    ) -> tuple[int, int, TelegramID]:
         entity = await self.get_input_entity(source)
         events = []
         intents = []
         metas = []
-        state_events = []
-        do_batch_send = self.backfill_msc2716
-        added_members = (
-            await self._get_members_at(prev_event_id)
-            if not self.bridge.homeserver_software.is_hungry and do_batch_send
-            else set()
-        )
-        before_first_msg_timestamp = 0
         tg_space = self.tgid if self.peer_type == "channel" else source.tgid
 
-        async def add_member(intent: IntentAPI, displayname: str, avatar_url: ContentURI) -> None:
-            if self.bridge.homeserver_software.is_hungry or intent.mxid in added_members:
-                return
-            added_members.add(intent.mxid)
-            if not do_batch_send:
-                # TODO leave these members?
-                await intent.ensure_joined(self.mxid)
-                return
-            invite_event = BatchSendStateEvent(
-                type=EventType.ROOM_MEMBER,
-                state_key=intent.mxid,
-                sender=self.main_intent.mxid,
-                timestamp=before_first_msg_timestamp,
-                content=MemberStateEventContent(
-                    membership=Membership.INVITE,
-                    displayname=displayname,
-                    avatar_url=avatar_url,
-                ),
-            )
-            join_event = attr.evolve(
-                invite_event,
-                content=attr.evolve(invite_event.content, membership=Membership.JOIN),
-                sender=intent.mxid,
-            )
-            state_events.append(invite_event)
-            state_events.append(join_event)
-
         lowest_id = 0
+        first_id_found = False
         first_id = anchor_id
         message_count = 0
         minmax = {"min_id": anchor_id} if forward else {"max_id": anchor_id}
@@ -3149,11 +3099,11 @@ class Portal(DBPortal, BasePortal):
                 continue
             if not lowest_id or msg.id < lowest_id:
                 lowest_id = msg.id
-            if not before_first_msg_timestamp:
+            if not first_id_found:
                 first_id = msg.id
-                before_first_msg_timestamp = int(msg.date.timestamp() * 1000) - 1
+                first_id_found = True
 
-            converted, intent = await self._convert_batch_msg(source, client, msg, add_member)
+            converted, intent = await self._convert_batch_msg(source, client, msg)
             if converted is None:
                 continue
             d_event_id = None
@@ -3172,28 +3122,21 @@ class Portal(DBPortal, BasePortal):
                 f"Didn't get any events to send out of {message_count} messages fetched "
                 f"(first received ID: {first_id}, lowest: {lowest_id})"
             )
-            return None, 0, message_count, lowest_id
+            return 0, message_count, lowest_id
         self.log.debug(
             f"Got {len(events)} events to send out of {message_count} messages fetched "
             f"(first received ID: {first_id}, lowest: {lowest_id})"
         )
-        if do_batch_send:
-            resp = await self.main_intent.batch_send(
+        if self._enable_batch_sending:
+            resp = await self.main_intent.beeper_batch_send(
                 self.mxid,
-                prev_event_id,
-                batch_id=self.next_batch_id if not forward else None,
                 # We iterated the events in reverse chronological order,
                 # so reverse them before sending
                 events=list(reversed(events)),
-                state_events_at_start=state_events,
-                beeper_new_messages=forward,
+                forward=forward,
             )
-            if prev_event_id == self.first_event_id and resp.next_batch_id:
-                self.next_batch_id = resp.next_batch_id
-            base_insertion_event_id = resp.base_insertion_event_id
             event_ids = resp.event_ids
         else:
-            base_insertion_event_id = None
             event_ids = [
                 await intent.send_message_event(
                     self.mxid, evt.type, evt.content, timestamp=evt.timestamp
@@ -3218,16 +3161,20 @@ class Portal(DBPortal, BasePortal):
                 if msg is not None
             ]
         )
-        return base_insertion_event_id, len(events), message_count, lowest_id
+        return len(events), message_count, lowest_id
 
     def _split_dm_reaction_counts(self, counts: list[ReactionCount]) -> list[MessagePeerReaction]:
         reactions = []
         for item in counts:
             if item.count == 2:
                 reactions += [
-                    MessagePeerReaction(reaction=item.reaction, peer_id=PeerUser(self.tgid)),
                     MessagePeerReaction(
-                        reaction=item.reaction, peer_id=PeerUser(self.tg_receiver)
+                        reaction=item.reaction, peer_id=PeerUser(self.tgid), date=None
+                    ),
+                    MessagePeerReaction(
+                        reaction=item.reaction,
+                        peer_id=PeerUser(self.tg_receiver),
+                        date=None,
                     ),
                 ]
             elif item.count == 1:
@@ -3235,6 +3182,7 @@ class Portal(DBPortal, BasePortal):
                     MessagePeerReaction(
                         reaction=item.reaction,
                         peer_id=PeerUser(self.tg_receiver if item.chosen_order else self.tgid),
+                        date=None,
                     )
                 )
         return reactions
@@ -3313,12 +3261,40 @@ class Portal(DBPortal, BasePortal):
                 recent_reactions = resp.reactions
 
         async with self.reaction_lock(dbm.mxid):
-            await self._handle_telegram_reactions_locked(
+            await self._handle_telegram_user_reactions_locked(
                 source, dbm, recent_reactions, total_count, timestamp=timestamp
             )
 
+    async def handle_telegram_bot_reactions(
+        self, source: au.AbstractUser, update: UpdateBotMessageReaction
+    ) -> None:
+        tg_space = self.tgid if self.peer_type == "channel" else source.tgid
+        dbm = await DBMessage.get_one_by_tgid(TelegramID(update.msg_id), tg_space)
+        if dbm is None:
+            return
+        reactions: dict[TelegramID, list[WrappedReaction]] = {}
+        custom_emoji_ids: list[int] = []
+        if isinstance(update.actor, PeerUser):
+            user_id = TelegramID(update.actor.user_id)
+        elif isinstance(update.actor, PeerChannel):
+            user_id = TelegramID(update.actor.channel_id)
+        else:
+            return
+        for reaction in update.new_reactions:
+            reactions.setdefault(user_id, []).append(WrappedReaction(reaction=reaction, date=None))
+        async with self.reaction_lock(dbm.mxid):
+            await self._handle_telegram_parsed_reactions_locked(
+                source,
+                dbm,
+                reactions,
+                custom_emoji_ids,
+                is_full=True,
+                only_user_id=user_id,
+                timestamp=update.date,
+            )
+
     @staticmethod
-    def _reactions_filter(lst: list[MessagePeerReaction], existing: DBReaction) -> bool:
+    def _reactions_filter(lst: list[WrappedReaction], existing: DBReaction) -> bool:
         if not lst:
             return False
         for wrapped_reaction in lst:
@@ -3341,7 +3317,7 @@ class Portal(DBPortal, BasePortal):
             return await source.get_max_reactions(is_premium)
         return 3 if is_premium else 1
 
-    async def _handle_telegram_reactions_locked(
+    async def _handle_telegram_user_reactions_locked(
         self,
         source: au.AbstractUser,
         msg: DBMessage,
@@ -3349,17 +3325,38 @@ class Portal(DBPortal, BasePortal):
         total_count: int,
         timestamp: datetime | None = None,
     ) -> None:
-        reactions: dict[TelegramID, list[MessagePeerReaction]] = {}
+        reactions: dict[TelegramID, list[WrappedReaction]] = {}
         custom_emoji_ids: list[int] = []
         for reaction in reaction_list:
             if isinstance(reaction.peer_id, (PeerUser, PeerChannel)) and isinstance(
                 reaction.reaction, (ReactionEmoji, ReactionCustomEmoji)
             ):
                 sender_user_id = p.Puppet.get_id_from_peer(reaction.peer_id)
-                reactions.setdefault(sender_user_id, []).append(reaction)
+                reactions.setdefault(sender_user_id, []).append(
+                    WrappedReaction(reaction.reaction, reaction.date)
+                )
                 if isinstance(reaction.reaction, ReactionCustomEmoji):
                     custom_emoji_ids.append(reaction.reaction.document_id)
         is_full = len(reaction_list) == total_count
+        await self._handle_telegram_parsed_reactions_locked(
+            source,
+            msg,
+            reactions,
+            custom_emoji_ids,
+            is_full=is_full,
+            timestamp=timestamp,
+        )
+
+    async def _handle_telegram_parsed_reactions_locked(
+        self,
+        source: au.AbstractUser,
+        msg: DBMessage,
+        reactions: dict[TelegramID, list[WrappedReaction]],
+        custom_emoji_ids: list[int],
+        is_full: bool,
+        only_user_id: TelegramID | None = None,
+        timestamp: datetime | None = None,
+    ) -> None:
         custom_emojis = await util.transfer_custom_emojis_to_matrix(source, custom_emoji_ids)
 
         existing_reactions = await DBReaction.get_all_by_message(msg.mxid, msg.mx_room)
@@ -3367,6 +3364,8 @@ class Portal(DBPortal, BasePortal):
         removed: list[DBReaction] = []
         for existing_reaction in existing_reactions:
             sender_id = existing_reaction.tg_sender
+            if only_user_id is not None and sender_id != only_user_id:
+                continue
             new_reactions = reactions.get(sender_id)
             if self._reactions_filter(new_reactions, existing_reaction):
                 if new_reactions is not None and len(new_reactions) == 0:
@@ -3444,6 +3443,8 @@ class Portal(DBPortal, BasePortal):
         self, source: au.AbstractUser, sender: p.Puppet | None, evt: Message
     ) -> None:
         if not self.mxid:
+            if source.is_relaybot and self.config["bridge.relaybot.ignore_unbridged_group_chat"]:
+                return
             self.log.debug("Got telegram message %d, but no room exists, creating...", evt.id)
             await self.create_matrix_room(source, invites=[source.mxid], update_if_exists=False)
             if not self.mxid:
@@ -3530,7 +3531,7 @@ class Portal(DBPortal, BasePortal):
         else:
             intent = self.main_intent
         is_bot = sender.is_bot if sender else False
-        converted = await self._msg_conv.convert(source, intent, is_bot, evt)
+        converted = await self._msg_conv.convert(source, intent, is_bot, self.is_channel, evt)
         if not converted:
             return
         await intent.set_typing(self.mxid, timeout=0)
@@ -3598,7 +3599,9 @@ class Portal(DBPortal, BasePortal):
     async def _mark_disappearing(
         self, event_id: EventID, seconds: int, expires_at: int | None
     ) -> None:
-        dm = DisappearingMessage(self.mxid, event_id, seconds, expiration_ts=expires_at * 1000)
+        dm = DisappearingMessage(
+            self.mxid, event_id, seconds, expiration_ts=expires_at * 1000 if expires_at else None
+        )
         await dm.insert()
         if expires_at:
             background_task.create(self._disappear_event(dm))
@@ -3606,7 +3609,7 @@ class Portal(DBPortal, BasePortal):
     async def _create_room_on_action(
         self, source: au.AbstractUser, action: TypeMessageAction
     ) -> bool:
-        if source.is_relaybot and self.config["bridge.ignore_unbridged_group_chat"]:
+        if source.is_relaybot and self.config["bridge.relaybot.ignore_unbridged_group_chat"]:
             return False
         create_and_exit = (MessageActionChatCreate, MessageActionChannelCreate)
         create_and_continue = (
@@ -3682,7 +3685,7 @@ class Portal(DBPortal, BasePortal):
                 end_reason = "disconnected"
             body = f"{call_type} {end_reason}"
             if action.duration:
-                body += f" ({format_duration(action.duration)}"
+                body += f" ({format_duration(action.duration)})"
             await self._send_message(
                 sender.intent_for(self),
                 TextMessageEventContent(msgtype=MessageType.NOTICE, body=body),
@@ -3710,6 +3713,18 @@ class Portal(DBPortal, BasePortal):
                     ),
                 ),
             )
+        elif isinstance(action, MessageActionBoostApply):
+            await self._send_message(
+                sender.intent_for(self),
+                TextMessageEventContent(
+                    msgtype=MessageType.EMOTE,
+                    body=(
+                        "boosted the group"
+                        if action.boosts == 1
+                        else f"boosted the group {action.boosts} times"
+                    ),
+                ),
+            )
         elif isinstance(action, MessageActionGameScore):
             # TODO handle game score
             pass

mautrix_telegram/portal_util/message_convert.py

@@ -34,6 +34,8 @@ from telethon.tl.types import (
     DocumentAttributeVideo,
     Game,
     InputPhotoFileLocation,
+    InputStickerSetID,
+    InputStickerSetShortName,
     Message,
     MessageEntityPre,
     MessageMediaContact,
@@ -42,12 +44,16 @@ from telethon.tl.types import (
     MessageMediaGame,
     MessageMediaGeo,
     MessageMediaGeoLive,
+    MessageMediaInvoice,
     MessageMediaPhoto,
     MessageMediaPoll,
+    MessageMediaStory,
     MessageMediaUnsupported,
     MessageMediaVenue,
     MessageMediaWebPage,
+    MessageReplyStoryHeader,
     PeerChannel,
+    PeerChat,
     PeerUser,
     Photo,
     PhotoCachedSize,
@@ -58,6 +64,8 @@ from telethon.tl.types import (
     Poll,
     TypeDocumentAttribute,
     TypePhotoSize,
+    UpdateShortChatMessage,
+    UpdateShortMessage,
     WebPage,
 )
 from telethon.utils import decode_waveform
@@ -104,6 +112,7 @@ class DocAttrs(NamedTuple):
     mime_type: str | None
     is_sticker: bool
     sticker_alt: str | None
+    sticker_pack_ref: dict | None
     width: int
     height: int
     is_gif: bool
@@ -142,6 +151,8 @@ class TelegramMessageConverter:
             MessageMediaUnsupported: self._convert_unsupported,
             MessageMediaGame: self._convert_game,
             MessageMediaContact: self._convert_contact,
+            MessageMediaStory: self._convert_story,
+            MessageMediaInvoice: self._convert_invoice,
         }
         self._allowed_media = tuple(self._media_converters.keys())
 
@@ -150,6 +161,7 @@ class TelegramMessageConverter:
         source: au.AbstractUser,
         intent: IntentAPI,
         is_bot: bool,
+        is_channel: bool,
         evt: Message,
         no_reply_fallback: bool = False,
         deterministic_reply_id: bool = False,
@@ -158,8 +170,13 @@ class TelegramMessageConverter:
         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, client=client)
+            if self._should_convert_full_document(evt.media, is_bot, is_channel):
+                convert_media = self._media_converters[type(evt.media)]
+                converted = await convert_media(
+                    source=source, intent=intent, evt=evt, client=client
+                )
+            else:
+                converted = await self._convert_document_thumb_only(source, intent, evt, client)
         elif evt.message:
             converted = await self._convert_text(source, intent, is_bot, evt, client)
         else:
@@ -192,6 +209,16 @@ class TelegramMessageConverter:
             )
         return converted
 
+    def _should_convert_full_document(self, media, is_bot: bool, is_channel: bool) -> bool:
+        if not isinstance(media, MessageMediaDocument):
+            return True
+        size = media.document.size
+        if is_bot and self.config["bridge.document_as_link_size.bot"]:
+            return size < self.config["bridge.document_as_link_size.bot"] * 1000**2
+        if is_channel and self.config["bridge.document_as_link_size.channel"]:
+            return size < self.config["bridge.document_as_link_size.channel"] * 1000**2
+        return True
+
     @staticmethod
     def _caption_to_message(converted: ConvertedMessage) -> None:
         content, caption = converted.content, converted.caption
@@ -252,21 +279,56 @@ class TelegramMessageConverter:
     ) -> None:
         if not evt.reply_to:
             return
+        elif isinstance(evt.reply_to, MessageReplyStoryHeader):
+            return
+
+        if evt.reply_to.quote and content.msgtype.is_text:
+            content.ensure_has_html()
+            quote_html = await formatter.telegram_text_to_matrix_html(
+                source, evt.reply_to.quote_text, evt.reply_to.quote_entities
+            )
+            content.formatted_body = (
+                f"<blockquote data-telegram-partial-reply>{quote_html}</blockquote>"
+                f"{content.formatted_body}"
+            )
+
         space = (
             evt.peer_id.channel_id
             if isinstance(evt, Message) and isinstance(evt.peer_id, PeerChannel)
             else source.tgid
         )
+        if isinstance(evt, Message):
+            evt_peer_id = evt.peer_id
+        elif isinstance(evt, UpdateShortMessage):
+            evt_peer_id = PeerUser(evt.user_id)
+        elif isinstance(evt, UpdateShortChatMessage):
+            evt_peer_id = PeerChat(evt.chat_id)
+        else:
+            evt_peer_id = None
+        if evt.reply_to.reply_to_peer_id and evt.reply_to.reply_to_peer_id != evt_peer_id:
+            if not self.config["bridge.cross_room_replies"]:
+                return
+            space = (
+                evt.reply_to.reply_to_peer_id.channel_id
+                if isinstance(evt.reply_to.reply_to_peer_id, PeerChannel)
+                else source.tgid
+            )
+
         reply_to_id = TelegramID(evt.reply_to.reply_to_msg_id)
         msg = await DBMessage.get_one_by_tgid(reply_to_id, space)
         no_fallback = no_fallback or self.config["bridge.disable_reply_fallbacks"]
-        if not msg or msg.mx_room != self.portal.mxid:
+        if not msg:
+            # TODO try to find room ID when generating deterministic ID for cross-room reply
             if deterministic_id:
                 content.set_reply(self.deterministic_event_id(space, reply_to_id))
             return
+        elif msg.mx_room != self.portal.mxid and not self.config["bridge.cross_room_replies"]:
+            return
         elif not isinstance(content, TextMessageEventContent) or no_fallback:
             # Not a text message, just set the reply metadata and return
             content.set_reply(msg.mxid)
+            if msg.mx_room != self.portal.mxid:
+                content.relates_to.in_reply_to["room_id"] = msg.mx_room
             return
 
         # Text message, try to fetch original message to generate reply fallback.
@@ -281,6 +343,8 @@ class TelegramMessageConverter:
         except Exception:
             self.log.exception("Failed to get event to add reply fallback")
             content.set_reply(msg.mxid)
+        if msg.mx_room != self.portal.mxid:
+            content.relates_to.in_reply_to["room_id"] = msg.mx_room
 
     @staticmethod
     def _photo_size_key(photo: TypePhotoSize) -> int:
@@ -429,7 +493,104 @@ class TelegramMessageConverter:
         return ConvertedMessage(
             content=content,
             caption=caption_content,
-            disappear_seconds=media.ttl_seconds,
+            disappear_seconds=self._adjust_ttl(media.ttl_seconds),
+        )
+
+    @staticmethod
+    def _adjust_ttl(ttl: int | None) -> int | None:
+        if not ttl:
+            return None
+        elif ttl == 2147483647:
+            # View-once media, set low TTL
+            return 15
+        else:
+            # Increase media TTL because it's supposed to be counted from opening the media,
+            # but we can only count it from read receipt.
+            return ttl * 5
+
+    async def _convert_document_thumb_only(
+        self,
+        source: au.AbstractUser,
+        intent: IntentAPI,
+        evt: Message,
+        client: MautrixTelegramClient,
+    ) -> ConvertedMessage | None:
+        document = evt.media.document
+
+        if not document:
+            return None
+
+        external_link_content = "Unsupported file, please access directly on Telegram"
+
+        external_url = self._get_external_url(evt)
+        # We don't generate external URLs for bot users so only set if known
+        if external_url is not None:
+            external_link_content = (
+                f"Unsupported file, please access directly on Telegram here: {external_url}"
+            )
+
+        attrs = _parse_document_attributes(document.attributes)
+        file = None
+
+        thumb_loc, thumb_size = self.get_largest_photo_size(document)
+        if thumb_size and not isinstance(thumb_size, (PhotoSize, PhotoCachedSize)):
+            self.log.debug(f"Unsupported thumbnail type {type(thumb_size)}")
+            thumb_loc = None
+            thumb_size = None
+        if thumb_loc:
+            try:
+                file = await util.transfer_thumbnail_to_matrix(
+                    client,
+                    intent,
+                    thumb_loc,
+                    video=None,
+                    mime_type=document.mime_type,
+                    encrypt=self.portal.encrypted,
+                    async_upload=self.config["homeserver.async_media"],
+                )
+            except Exception:
+                self.log.exception("Failed to transfer thumbnail")
+        if not file:
+            name = attrs.name or ""
+            caption = f"\n{evt.message}" if evt.message else ""
+            return ConvertedMessage(
+                content=TextMessageEventContent(
+                    msgtype=MessageType.NOTICE,
+                    body=f"{name}{caption}\n{external_link_content}",
+                )
+            )
+
+        info, name = _parse_document_meta(evt, file, attrs, thumb_size)
+
+        event_type = EventType.ROOM_MESSAGE
+        if not name:
+            ext = sane_mimetypes.guess_extension(file.mime_type) or ""
+            name = "unnamed_file" + ext
+
+        content = MediaMessageEventContent(
+            body=name,
+            info=info,
+            msgtype={
+                "video/": MessageType.VIDEO,
+                "audio/": MessageType.AUDIO,
+                "image/": MessageType.IMAGE,
+            }.get(info.mimetype[:6], MessageType.FILE),
+        )
+        if file.decryption_info:
+            content.file = file.decryption_info
+        else:
+            content.url = file.mxc
+
+        caption_content = (
+            await formatter.telegram_to_matrix(evt, source, client) if evt.message else None
+        )
+        caption_content = f"{caption_content}\n{external_link_content}"
+
+        return ConvertedMessage(
+            type=event_type,
+            content=content,
+            caption=caption_content,
+            disappear_seconds=self._adjust_ttl(evt.media.ttl_seconds),
         )
 
     async def _convert_document(
@@ -441,6 +602,9 @@ class TelegramMessageConverter:
     ) -> ConvertedMessage | None:
         document = evt.media.document
 
+        if not document:
+            return None
+
         attrs = _parse_document_attributes(document.attributes)
 
         if document.size > self.matrix.media_config.upload_size:
@@ -538,7 +702,7 @@ class TelegramMessageConverter:
             type=event_type,
             content=content,
             caption=caption_content,
-            disappear_seconds=evt.media.ttl_seconds,
+            disappear_seconds=self._adjust_ttl(evt.media.ttl_seconds),
         )
 
     @staticmethod
@@ -700,10 +864,33 @@ class TelegramMessageConverter:
             )
         return ConvertedMessage(content=content)
 
+    @staticmethod
+    async def _convert_story(
+        source: au.AbstractUser, evt: Message, client: MautrixTelegramClient, **_
+    ) -> ConvertedMessage:
+        content = await formatter.telegram_to_matrix(
+            evt, source, client, override_text="Stories are not yet supported"
+        )
+        content.msgtype = MessageType.NOTICE
+        content["fi.mau.telegram.unsupported"] = True
+        return ConvertedMessage(content=content)
+
+    @staticmethod
+    async def _convert_invoice(
+        source: au.AbstractUser, evt: Message, client: MautrixTelegramClient, **_
+    ) -> ConvertedMessage:
+        content = await formatter.telegram_to_matrix(
+            evt, source, client, override_text="Invoices are not yet supported"
+        )
+        content.msgtype = MessageType.NOTICE
+        content["fi.mau.telegram.unsupported"] = True
+        return ConvertedMessage(content=content)
+
 
 def _parse_document_attributes(attributes: list[TypeDocumentAttribute]) -> DocAttrs:
     name, mime_type, is_sticker, sticker_alt, width, height = None, None, False, None, 0, 0
     is_gif, is_audio, is_voice, duration, waveform = False, False, False, 0, bytes()
+    sticker_pack_ref = None
     for attr in attributes:
         if isinstance(attr, DocumentAttributeFilename):
             name = name or attr.file_name
@@ -711,6 +898,13 @@ def _parse_document_attributes(attributes: list[TypeDocumentAttribute]) -> DocAt
         elif isinstance(attr, DocumentAttributeSticker):
             is_sticker = True
             sticker_alt = attr.alt
+            if isinstance(attr.stickerset, InputStickerSetID):
+                sticker_pack_ref = {
+                    "id": str(attr.stickerset.id),
+                    "access_hash": str(attr.stickerset.access_hash),
+                }
+            elif isinstance(attr.stickerset, InputStickerSetShortName):
+                sticker_pack_ref = {"short_name": attr.stickerset.short_name}
         elif isinstance(attr, DocumentAttributeAnimated):
             is_gif = True
         elif isinstance(attr, DocumentAttributeVideo):
@@ -724,17 +918,18 @@ def _parse_document_attributes(attributes: list[TypeDocumentAttribute]) -> DocAt
             waveform = decode_waveform(attr.waveform) if attr.waveform else b""
 
     return DocAttrs(
-        name,
-        mime_type,
-        is_sticker,
-        sticker_alt,
-        width,
-        height,
-        is_gif,
-        is_audio,
-        is_voice,
-        duration,
-        waveform,
+        name=name,
+        mime_type=mime_type,
+        is_sticker=is_sticker,
+        sticker_alt=sticker_alt,
+        sticker_pack_ref=sticker_pack_ref,
+        width=width,
+        height=height,
+        is_gif=is_gif,
+        is_audio=is_audio,
+        is_voice=is_voice,
+        duration=duration,
+        waveform=waveform,
     )
 
 
@@ -760,6 +955,13 @@ def _parse_document_meta(
         mime_type = file.mime_type or document.mime_type
     info = ImageInfo(size=file.size, mimetype=mime_type)
 
+    if attrs.is_sticker:
+        info["fi.mau.telegram.sticker"] = {
+            "alt": attrs.sticker_alt,
+            "id": str(document.id),
+            "pack": attrs.sticker_pack_ref,
+        }
+
     if attrs.mime_type and not file.was_converted:
         file.mime_type = attrs.mime_type or file.mime_type
     if file.width and file.height:
@@ -779,6 +981,10 @@ def _parse_document_meta(
             size=file.thumbnail.size,
         )
     elif attrs.is_sticker:
+        if not info.width or not info.height:
+            info.width = 256
+            info.height = 256
+
         # This is a hack for bad clients like Element iOS that require a thumbnail
         info.thumbnail_info = ImageInfo.deserialize(info.serialize())
         if file.decryption_info:

mautrix_telegram/portal_util/power_levels.py

@@ -83,9 +83,11 @@ def get_base_power_levels(
         levels.users_default = overrides.get("users_default", 0)
         levels.events_default = overrides.get(
             "events_default",
-            50
-            if portal.peer_type == "channel" and not portal.megagroup or dbr.send_messages
-            else 0,
+            (
+                50
+                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

mautrix_telegram/puppet.py

@@ -269,11 +269,14 @@ class Puppet(DBPuppet, BasePuppet):
             is_bot != self.is_bot or is_channel != self.is_channel or is_premium != self.is_premium
         )
 
-        self.is_bot = is_bot
+        if is_bot is not None:
+            self.is_bot = is_bot
         self.is_channel = is_channel
-        self.is_premium = is_premium
+        if is_premium is not None:
+            self.is_premium = is_premium
 
-        if self.username != info.username:
+        if self.username != info.username and (info.username or not info.min):
+            self.log.debug(f"Updating username {self.username} -> {info.username}")
             self.username = info.username
             changed = True
 

mautrix_telegram/tgclient.py

@@ -22,6 +22,7 @@ from telethon.tl.patched import Message
 from telethon.tl.types import (
     InputMediaUploadedDocument,
     InputMediaUploadedPhoto,
+    InputReplyToMessage,
     TypeDocumentAttribute,
     TypeInputMedia,
     TypeInputPeer,
@@ -67,6 +68,10 @@ class MautrixTelegramClient(TelegramClient):
         entity = await self.get_input_entity(entity)
         reply_to = utils.get_message_id(reply_to)
         request = SendMediaRequest(
-            entity, media, message=caption or "", entities=entities or [], reply_to_msg_id=reply_to
+            entity,
+            media,
+            message=caption or "",
+            entities=entities or [],
+            reply_to=InputReplyToMessage(reply_to_msg_id=reply_to) if reply_to else None,
         )
         return self._get_response_message(request, await self(request), entity)

mautrix_telegram/user.py

@@ -23,6 +23,7 @@ import time
 from telethon.errors import (
     AuthKeyDuplicatedError,
     AuthKeyError,
+    AuthKeyNotFound,
     RPCError,
     TakeoutInitDelayError,
     UnauthorizedError,
@@ -215,7 +216,7 @@ class User(DBUser, AbstractUser, BaseUser):
         async with self._ensure_started_lock:
             return cast(User, await super().ensure_started(even_if_no_session))
 
-    async def on_signed_out(self, err: UnauthorizedError | AuthKeyError) -> None:
+    async def on_signed_out(self, err: UnauthorizedError | AuthKeyError | AuthKeyNotFound) -> None:
         error_code = "tg-auth-error"
         if isinstance(err, AuthKeyDuplicatedError):
             error_code = "tg-auth-key-duplicated"
@@ -236,8 +237,8 @@ class User(DBUser, AbstractUser, BaseUser):
     async def start(self, delete_unless_authenticated: bool = False) -> User:
         try:
             await super().start()
-        except AuthKeyDuplicatedError as e:
-            self.log.warning("Got AuthKeyDuplicatedError in start()")
+        except (AuthKeyDuplicatedError, AuthKeyNotFound) as e:
+            self.log.warning(f"Got {type(e).__name__} in start()")
             await self.on_signed_out(e)
             if not delete_unless_authenticated:
                 # The caller wants the client to be connected, so restart the connection.
@@ -297,9 +298,11 @@ class User(DBUser, AbstractUser, BaseUser):
             self._track_metric(METRIC_CONNECTED, connected)
             if connected:
                 await self.push_bridge_state(
-                    BridgeStateEvent.BACKFILLING
-                    if self._is_backfilling
-                    else BridgeStateEvent.CONNECTED,
+                    (
+                        BridgeStateEvent.BACKFILLING
+                        if self._is_backfilling
+                        else BridgeStateEvent.CONNECTED
+                    ),
                     info=self._bridge_state_info,
                 )
             else:
@@ -617,8 +620,11 @@ class User(DBUser, AbstractUser, BaseUser):
             await self.stop()
             await sess.delete()
 
+        # Drop LOGGED_OUT states if the user was already logged out previously
+        # and doesn't have a remote ID anymore
         # TODO send a management room notice for non-manual logouts?
-        await self.push_bridge_state(state, error=error, message=message)
+        if self.tgid or state != BridgeStateEvent.LOGGED_OUT:
+            await self.push_bridge_state(state, error=error, message=message)
         if delete:
             await self.delete()
             self.by_mxid.pop(self.mxid, None)
@@ -974,11 +980,18 @@ class User(DBUser, AbstractUser, BaseUser):
         self.log.debug("Contact syncing complete")
         return contacts
 
+    @property
+    def _available_reactions_up_to_date(self) -> bool:
+        return (
+            bool(self._available_emoji_reactions)
+            and self._available_emoji_reactions_fetched + 12 * 60 * 60 > time.monotonic()
+        )
+
     async def get_available_reactions(self) -> set[str]:
-        if self._available_emoji_reactions_fetched + 12 * 60 * 60 > time.monotonic():
+        if self._available_reactions_up_to_date:
             return self._available_emoji_reactions
         async with self._available_emoji_reactions_lock:
-            if self._available_emoji_reactions_fetched + 12 * 60 * 60 > time.monotonic():
+            if self._available_reactions_up_to_date:
                 return self._available_emoji_reactions
             self.log.debug("Fetching available emoji reactions")
             available_reactions = await self.client(
@@ -988,13 +1001,18 @@ class User(DBUser, AbstractUser, BaseUser):
                 self._available_emoji_reactions = {
                     react.reaction
                     for react in available_reactions.reactions
-                    if self.is_premium or not react.premium
+                    if not react.inactive and (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
                 )
+            elif self._available_emoji_reactions is None:
+                self.log.warning(
+                    f"Got {available_reactions} in response to available reactions request"
+                    " even though nothing is cached"
+                )
             return self._available_emoji_reactions
 
     def tl_to_json(self) -> Any:

mautrix_telegram/util/__init__.py

@@ -4,6 +4,7 @@ from .file_transfer import (
     convert_image,
     transfer_custom_emojis_to_matrix,
     transfer_file_to_matrix,
+    transfer_thumbnail_to_matrix,
     unicode_custom_emoji_map,
 )
 from .parallel_file_transfer import parallel_transfer_to_telegram

mautrix_telegram/web/provisioning/__init__.py

@@ -130,9 +130,9 @@ class ProvisioningAPI(AuthAPI):
                 "about": portal.about,
                 "username": portal.username,
                 "megagroup": portal.megagroup,
-                "can_unbridge": (await portal.can_user_perform(user, "unbridge"))
-                if user
-                else False,
+                "can_unbridge": (
+                    (await portal.can_user_perform(user, "unbridge")) if user else False
+                ),
             }
         )
 
@@ -188,9 +188,11 @@ class ProvisioningAPI(AuthAPI):
             if force in ("delete", "unbridge"):
                 delete = force == "delete"
                 await portal.cleanup_portal(
-                    "Portal deleted (moving to another room)"
-                    if delete
-                    else "Room unbridged (portal moving to another room)",
+                    (
+                        "Portal deleted (moving to another room)"
+                        if delete
+                        else "Room unbridged (portal moving to another room)"
+                    ),
                     puppets_only=not delete,
                 )
             else:

optional-requirements.txt

@@ -7,14 +7,14 @@ aiodns
 brotli
 
 #/qr_login
-pillow>=4,<10
+pillow>=10.0.1,<11
 qrcode>=6,<8
 
 #/formattednumbers
 phonenumbers>=8,<9
 
 #/metrics
-prometheus_client>=0.6,<0.18
+prometheus_client>=0.6,<0.21
 
 #/e2be
 python-olm>=3,<4
@@ -22,7 +22,7 @@ pycryptodome>=3,<4
 unpaddedbase64>=1,<3
 
 #/sqlite
-aiosqlite>=0.16,<0.20
+aiosqlite>=0.16,<0.21
 
 #/proxy
 python-socks[asyncio]

pyproject.toml

@@ -9,4 +9,4 @@ line_length = 99
 
 [tool.black]
 line-length = 99
-target-version = ["py38"]
+target-version = ["py310"]

requirements.txt

@@ -1,10 +1,10 @@
-ruamel.yaml>=0.15.35,<0.18
+ruamel.yaml>=0.15.35,<0.19
 python-magic>=0.4,<0.5
 commonmark>=0.8,<0.10
 aiohttp>=3,<4
 yarl>=1,<2
-mautrix>=0.20.0,<0.21
-tulir-telethon==1.29.0a2
-asyncpg>=0.20,<0.28
+mautrix>=0.20.6,<0.21
+tulir-telethon==1.37.0a1
+asyncpg>=0.20,<0.30
 mako>=1,<2
 setuptools

setup.py

@@ -51,7 +51,7 @@ setuptools.setup(
 
     install_requires=install_requires,
     extras_require=extras_require,
-    python_requires="~=3.9",
+    python_requires="~=3.10",
 
     classifiers=[
         "Development Status :: 4 - Beta",
@@ -60,9 +60,9 @@ setuptools.setup(
         "Framework :: AsyncIO",
         "Programming Language :: Python",
         "Programming Language :: Python :: 3",
-        "Programming Language :: Python :: 3.9",
         "Programming Language :: Python :: 3.10",
         "Programming Language :: Python :: 3.11",
+        "Programming Language :: Python :: 3.12",
     ],
     package_data={"mautrix_telegram": [
         "web/public/*.mako", "web/public/*.png", "web/public/*.css",