Bump version to 0.11.3

async media: add ability to upload media asynchronously

.editorconfig

@@ -17,5 +17,5 @@ max_line_length = 99
 [*.{yaml,yml,py}]
 indent_style = space
 
-[{.gitlab-ci.yml,.pre-commit-config.yaml}]
+[{.gitlab-ci.yml,.pre-commit-config.yaml,mautrix_telegram/web/provisioning/spec.yaml}]
 indent_size = 2

.github/workflows/python-lint.yml

@@ -16,7 +16,7 @@ jobs:
     - uses: psf/black@stable
       with:
         src: "./mautrix_telegram"
-        version: "22.1.0"
+        version: "22.3.0"
     - name: pre-commit
       run: |
         pip install pre-commit

.pre-commit-config.yaml

@@ -7,17 +7,14 @@ repos:
       - id: end-of-file-fixer
       - id: check-yaml
       - id: check-added-large-files
-  # TODO convert to use the upstream psf/black when
-  # https://github.com/psf/black/issues/2493 gets fixed
-  - repo: local
+  - repo: https://github.com/psf/black
+    rev: 22.3.0
     hooks:
       - id: black
-        name: black
-        entry: black --check
-        language: system
-        files: ^mautrix_telegram/.*\.py$
+        language_version: python3
+        files: ^mautrix_telegram/.*\.pyi?$
   - repo: https://github.com/PyCQA/isort
     rev: 5.10.1
     hooks:
       - id: isort
-        files: ^mautrix_telegram/.*$
+        files: ^mautrix_telegram/.*\.pyi?$

CHANGELOG.md

@@ -1,3 +1,44 @@
+# v0.11.3 (2022-04-17)
+
+**N.B.** This release drops support for old homeservers which don't support the
+new `/v3` API endpoints. Synapse 1.48+, Dendrite 0.6.5+ and Conduit 0.4.0+ are
+supported. Legacy `r0` API support can be temporarily re-enabled with `pip install mautrix==0.16.0`.
+However, this option will not be available in future releases.
+
+### Added
+* Added `list-invite-links` command to list invite links in a chat.
+* Added option to use [MSC2246] async media uploads.
+* Provisioning API for listing contacts and starting private chats.
+
+### Improved
+* Dropped Python 3.7 support.
+* Telegram->Matrix message formatter will now replace `t.me/c/chatid/messageid`
+  style links with a link to the bridged Matrix event (in addition to the
+  previously supported `t.me/username/messageid` links).
+* Updated formatting converter to keep newlines in code blocks as `\n` instead
+  of converting them to `<br/>`.
+* Removed `max_document_size` option. The bridge will now fetch the max size
+  automatically using the media repo config endpoint.
+* Removed redundant `msgtype` field in sticker events sent to Matrix.
+* Disabled file logging in Docker image by default.
+  * If you want to enable it, set the `filename` in the file log handler to a
+    path that is writable, then add `"file"` back to `logging.root.handlers`.
+* Reactions are now marked as read when bridging read receipts from Matrix.
+
+### Fixed
+* Fixed `!tg bridge` throwing error if the parameter is not an integer
+* Fixed `!tg bridge` failing if the command had been previously run with an
+  incorrectly prefixed chat ID (e.g. `!tg bridge -1234567` followed by
+  `!tg bridge -1001234567`).
+* Fixed `bridge_matrix_leave` config option not actually being used correctly.
+* Fixed public channel mentions always bridging into a user mention on Matrix
+  rather than a room mention.
+  * The bridge will now make room mentions if the portal exists and fall back
+    to user mentions otherwise.
+* Fixed newlines being lost in unformatted forwarded messages.
+
+[MSC2246]: https://github.com/matrix-org/matrix-spec-proposals/pull/2246
+
 # v0.11.2 (2022-02-14)
 
 **N.B.** This will be the last release to support Python 3.7. Future versions
@@ -229,8 +270,8 @@ path.
 * Bridging events of a user whose power level is malformed (i.e. a string
   instead of an integer) now works.
 
-[MSC2409]: https://github.com/matrix-org/matrix-doc/pull/2409
-[MSC2778]: https://github.com/matrix-org/matrix-doc/pull/2778
+[MSC2409]: https://github.com/matrix-org/matrix-spec-proposals/pull/2409
+[MSC2778]: https://github.com/matrix-org/matrix-spec-proposals/pull/2778
 
 # v0.8.2 (2020-07-27)
 
@@ -278,7 +319,7 @@ update (v0.5.8) and a fix to the Docker image.
 * Fixed `sync_direct_chats` option creating non-working portals.
 * Fixed video thumbnailing sometimes leaving behind downloaded videos in `/tmp`.
 
-[MSC2346]: https://github.com/matrix-org/matrix-doc/pull/2346
+[MSC2346]: https://github.com/matrix-org/matrix-spec-proposals/pull/2346
 
 ## rc1 (2020-04-25)
 

dev-requirements.txt

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

docker-run.sh

@@ -2,7 +2,13 @@
 
 # Define functions.
 function fixperms {
-	chown -R $UID:$GID /data /opt/mautrix-telegram
+	chown -R $UID:$GID /data
+
+	# /opt/mautrix-telegram is read-only, so disable file logging if it's pointing there.
+	if [[ "$(yq e '.logging.handlers.file.filename' /data/config.yaml)" == "./mautrix-telegram.log" ]]; then
+		yq -I4 e -i 'del(.logging.root.handlers[] | select(. == "file"))' /data/config.yaml
+		yq -I4 e -i 'del(.logging.handlers.file)' /data/config.yaml
+	fi
 }
 
 cd /opt/mautrix-telegram
@@ -18,7 +24,10 @@ if [ ! -f /data/config.yaml ]; then
 fi
 
 if [ ! -f /data/registration.yaml ]; then
-	python3 -m mautrix_telegram -g -c /data/config.yaml -r /data/registration.yaml
+	python3 -m mautrix_telegram -g -c /data/config.yaml -r /data/registration.yaml || exit $?
+	echo "Didn't find a registration file."
+	echo "Generated one for you."
+	echo "See https://docs.mau.fi/bridges/general/registering-appservices.html on how to use it."
 	fixperms
 	exit
 fi

mautrix_telegram/__init__.py

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

mautrix_telegram/commands/handler.py

@@ -137,9 +137,9 @@ class CommandHandler(BaseCommandHandler):
 
     async def get_permission_error(self, evt: CommandEvent) -> str | None:
         if self.needs_puppeting and not evt.sender.puppet_whitelisted:
-            return "This command requires puppeting privileges."
+            return "That command is limited to users with puppeting privileges."
         elif self.needs_matrix_puppeting and not evt.sender.matrix_puppet_whitelisted:
-            return "This command requires Matrix puppeting privileges."
+            return "That command is limited to users with full puppeting privileges."
         return await super().get_permission_error(evt)
 
     def has_permission(self, key: HelpCacheKey) -> bool:

mautrix_telegram/commands/matrix_auth.py

@@ -81,5 +81,5 @@ async def enter_matrix_token(evt: CommandEvent) -> EventID:
     except InvalidAccessToken:
         return await evt.reply("Failed to verify access token.")
     return await evt.reply(
-        "Replaced your Telegram account's Matrix puppet with {puppet.custom_mxid}."
+        f"Replaced your Telegram account's Matrix puppet with {puppet.custom_mxid}."
     )

mautrix_telegram/commands/portal/bridge.py

@@ -59,17 +59,22 @@ async def bridge(evt: CommandEvent) -> EventID:
 
     # The /id bot command provides the prefixed ID, so we assume
     tgid_str = evt.args[0]
-    if tgid_str.startswith("-100"):
-        tgid = TelegramID(int(tgid_str[4:]))
-        peer_type = "channel"
-    elif tgid_str.startswith("-"):
-        tgid = TelegramID(-int(tgid_str))
-        peer_type = "chat"
-    else:
+    tgid = None
+    try:
+        if tgid_str.startswith("-100"):
+            tgid = TelegramID(int(tgid_str[4:]))
+            peer_type = "channel"
+        elif tgid_str.startswith("-"):
+            tgid = TelegramID(-int(tgid_str))
+            peer_type = "chat"
+    except ValueError:
+        # Invalid integer
+        pass
+    if not tgid:
         return await evt.reply(
             "That doesn't seem like a prefixed Telegram chat ID.\n\n"
-            "If you did not get the ID using the `/id` bot command, please "
-            "prefix channel IDs with `-100` and normal group IDs with `-`.\n\n"
+            "If you did not get the ID using the `/id` bot command, please prefix"
+            "channel/supergroup IDs with `-100` and non-super group IDs with `-`.\n\n"
             "Bridging private chats to existing rooms is not allowed."
         )
 
@@ -80,7 +85,7 @@ async def bridge(evt: CommandEvent) -> EventID:
             "If you're the bridge admin, try "
             "`$cmdprefix+sp filter whitelist <Telegram chat ID>` first."
         )
-    if portal.mxid:
+    elif portal.mxid:
         has_portal_message = (
             "That Telegram chat already has a portal at "
             f"[{portal.alias or portal.mxid}](https://matrix.to/#/{portal.mxid}). "
@@ -96,7 +101,7 @@ async def bridge(evt: CommandEvent) -> EventID:
             "mxid": portal.mxid,
             "bridge_to_mxid": room_id,
             "tgid": portal.tgid,
-            "peer_type": portal.peer_type,
+            "peer_type": peer_type,
             "force_use_bot": force_use_bot,
         }
         return await evt.reply(
@@ -112,7 +117,7 @@ async def bridge(evt: CommandEvent) -> EventID:
         "action": "Room bridging",
         "bridge_to_mxid": room_id,
         "tgid": portal.tgid,
-        "peer_type": portal.peer_type,
+        "peer_type": peer_type,
         "force_use_bot": force_use_bot,
     }
     return await evt.reply(
@@ -163,6 +168,18 @@ async def confirm_bridge(evt: CommandEvent) -> EventID | None:
     is_logged_in = await evt.sender.is_logged_in() and not status["force_use_bot"]
 
     if "mxid" in status:
+        if portal.peer_type != status["peer_type"]:
+            evt.log.warning(
+                "Portal %d in database has mismatching peer type %s (expected %s),"
+                " trusting database as a room already existed",
+                portal.tgid,
+                portal.peer_type,
+                status["peer_type"],
+            )
+            await evt.reply(
+                "Mismatching peer type in command and portal table, "
+                "trusting portal as room already existed"
+            )
         ok, coro = await cleanup_old_portal_while_bridging(evt, portal)
         if not ok:
             return None
@@ -181,6 +198,19 @@ async def confirm_bridge(evt: CommandEvent) -> EventID | None:
             "Please use `$cmdprefix+sp continue` to confirm the bridging or "
             "`$cmdprefix+sp cancel` to cancel."
         )
+    elif portal.peer_type != status["peer_type"]:
+        evt.log.warning(
+            "Portal %d in database has mismatching peer type %s (expected %s),"
+            " trusting new peer type as there's no existing room",
+            portal.tgid,
+            portal.peer_type,
+            status["peer_type"],
+        )
+        await evt.reply(
+            "Mismatching peer type in command and portal table, "
+            "trusting you as portal room doesn't exist"
+        )
+        portal.peer_type = status["peer_type"]
 
     evt.sender.command_status = None
     async with portal._room_create_lock:
@@ -221,7 +251,7 @@ async def _locked_confirm_bridge(
     await portal.save()
     await portal.update_bridge_info()
 
-    asyncio.create_task(portal.update_matrix_room(user, entity, direct=False, levels=levels))
+    asyncio.create_task(portal.update_matrix_room(user, entity, levels=levels))
 
     await warn_missing_power(levels, evt)
 

mautrix_telegram/commands/portal/misc.py

@@ -25,12 +25,22 @@ from telethon.errors import (
     UsernameNotModifiedError,
     UsernameOccupiedError,
 )
+from telethon.helpers import add_surrogate
 from telethon.tl.functions.channels import GetFullChannelRequest
-from telethon.tl.functions.messages import GetFullChatRequest
+from telethon.tl.functions.messages import GetExportedChatInvitesRequest, GetFullChatRequest
+from telethon.tl.types import (
+    ChatInviteExported,
+    InputMessageEntityMentionName,
+    InputUserSelf,
+    MessageEntityMention,
+    TypeInputPeer,
+    TypeInputUser,
+)
+from telethon.tl.types.messages import ExportedChatInvites
 
 from mautrix.types import EventID
 
-from ... import portal as po
+from ... import formatter as fmt, portal as po, puppet as pu
 from .. import SECTION_MISC, SECTION_PORTAL_MANAGEMENT, CommandEvent, command_handler
 from .util import user_has_power_level
 
@@ -101,30 +111,37 @@ async def get_id(evt: CommandEvent) -> EventID:
 
 
 invite_link_usage = (
-    "**Usage:** `$cmdprefix+sp invite-link [--uses=<amount>] [--expire=<delta>]`"
+    "**Usage:** `$cmdprefix+sp invite-link "
+    "[--uses=<amount>] [--expire=<delta>] [--request-needed] -- [title]`"
     "\n\n"
     "* `--uses`: the number of times the invite link can be used."
     "            Defaults to unlimited.\n"
     "* `--expire`: the duration after which the link will expire."
-    "              A number suffixed with d(ay), h(our), m(inute) or s(econd)"
+    "              A number suffixed with d(ay), h(our), m(inute) or s(econd)\n"
+    "* `--request-needed`: should the link require admins to approve joins?\n"
+    "* `title`: a description of the link (only shown to admins)."
 )
 
 
 def _parse_flag(args: list[str]) -> tuple[str, str]:
     arg = args.pop(0).lower()
+    if arg == "--":
+        return "", ""
+    value = ""
     if arg.startswith("--"):
-        value_start = arg.index("=")
-        if value_start:
+        value_start = arg.find("=")
+        if value_start > 0:
             flag = arg[2:value_start]
             value = arg[value_start + 1 :]
         else:
             flag = arg[2:]
-            value = args.pop(0).lower()
+            if arg not in ("request", "request-needed"):
+                value = args.pop(0).lower()
     elif arg.startswith("-"):
         flag = arg[1]
         if len(arg) > 3 and arg[2] == "=":
             value = arg[3:]
-        else:
+        elif arg != "r":
             value = args.pop(0).lower()
     else:
         raise ValueError("invalid flag")
@@ -159,18 +176,24 @@ def _parse_delta(value: str) -> timedelta | None:
 @command_handler(
     help_section=SECTION_PORTAL_MANAGEMENT,
     help_text="Get a Telegram invite link to the current chat.",
-    help_args="[--uses=<amount>] [--expire=<time delta, e.g. 1d>]",
+    help_args="[--uses=<amount>] [--expire=<time delta, e.g. 1d>] [--request-needed] -- [title]",
 )
 async def invite_link(evt: CommandEvent) -> EventID:
+    if not evt.is_portal:
+        return await evt.reply("This is not a portal room.")
+
     # TODO once we switch to Python 3.9 minimum, use argparse with exit_on_error=False
     uses = None
     expire = None
+    request_needed = False
     while evt.args:
         try:
             flag, value = _parse_flag(evt.args)
         except (ValueError, IndexError):
             return await evt.reply(invite_link_usage)
-        if flag in ("uses", "u"):
+        if not flag:
+            break
+        elif flag in ("uses", "u"):
             try:
                 uses = int(value)
             except ValueError:
@@ -180,23 +203,90 @@ async def invite_link(evt: CommandEvent) -> EventID:
             if not expire_delta:
                 await evt.reply("Invalid format for expiry time delta")
             expire = datetime.now() + expire_delta
+        elif flag in ("request", "request-needed", "r"):
+            request_needed = True
+    title = " ".join(evt.args)
 
-    portal = await po.Portal.get_by_mxid(evt.room_id)
-    if not portal:
-        return await evt.reply("This is not a portal room.")
-
-    if portal.peer_type == "user":
+    if evt.portal.peer_type == "user":
         return await evt.reply("You can't invite users to private chats.")
 
     try:
-        link = await portal.get_invite_link(evt.sender, uses=uses, expire=expire)
-        return await evt.reply(f"Invite link to {portal.title}: {link}")
+        link = await evt.portal.get_invite_link(
+            evt.sender, uses=uses, expire=expire, request_needed=request_needed, title=title
+        )
+        return await evt.reply(f"Invite link to {evt.portal.title}: {link}")
     except ValueError as e:
         return await evt.reply(e.args[0])
     except ChatAdminRequiredError:
         return await evt.reply("You don't have the permission to create an invite link.")
 
 
+async def _format_invite_link(link: ChatInviteExported) -> str:
+    desc = f"* {link.link}"
+    if link.title:
+        desc += f" - {link.title}"
+    if link.expire_date:
+        desc += f"  \n  Expires at {link.expire_date.isoformat()}"
+    if link.usage_limit:
+        desc += f"  \n  Used {link.usage or 0} out of {link.usage_limit} times"
+    elif link.usage:
+        desc += f"  \n  Used {link.usage} times"
+    else:
+        desc += "  \n  Never used"
+    if link.request_needed:
+        desc += "  \n  Join requests enabled - using link requires admin approval"
+    return desc
+
+
+async def _hacky_find_mention(evt: CommandEvent) -> TypeInputUser | TypeInputPeer | None:
+    if len(evt.args) == 0:
+        return None
+    text, entities = await fmt.matrix_to_telegram(
+        evt.sender.client, text=evt.content.body, html=evt.content.formatted_body
+    )
+    for entity in entities:
+        if isinstance(entity, MessageEntityMention):
+            admin_username = add_surrogate(text)[entity.offset + 1 : entity.offset + entity.length]
+            return await evt.sender.client.get_input_entity(admin_username)
+        elif isinstance(entity, InputMessageEntityMentionName):
+            return entity.user_id
+    return None
+
+
+@command_handler(
+    help_section=SECTION_PORTAL_MANAGEMENT,
+    help_text="List existing Telegram invite links to the current chat.",
+    help_args="[creator]",
+)
+async def list_invite_links(evt: CommandEvent) -> EventID:
+    admin_id = InputUserSelf()
+    try:
+        admin_id = await _hacky_find_mention(evt) or InputUserSelf()
+    except Exception:
+        pass
+    resp: ExportedChatInvites = await evt.sender.client(
+        GetExportedChatInvitesRequest(
+            peer=await evt.portal.get_input_entity(evt.sender),
+            admin_id=admin_id,
+            limit=100,
+        )
+    )
+    if resp.count == 0:
+        if isinstance(admin_id, InputUserSelf):
+            return await evt.reply("You haven't created any invite links to the current chat")
+        else:
+            return await evt.reply("That user hasn't created any invite links to the current chat")
+    formatted_links = "\n".join([await _format_invite_link(link) for link in resp.invites])
+    if isinstance(admin_id, InputUserSelf):
+        await evt.reply(f"Your links to this chat:\n\n{formatted_links}")
+    else:
+        puppet = await pu.Puppet.get_by_peer(admin_id)
+        await evt.reply(
+            f"[{puppet.displayname}](https://matrix.to/#/{puppet.mxid})'s links to this chat:\n\n"
+            f"{formatted_links}"
+        )
+
+
 @command_handler(
     help_section=SECTION_PORTAL_MANAGEMENT,
     help_text="Upgrade a normal Telegram group to a supergroup.",

mautrix_telegram/commands/telegram/misc.py

@@ -26,6 +26,7 @@ from telethon.errors import (
     EmoticonInvalidError,
     InviteHashExpiredError,
     InviteHashInvalidError,
+    InviteRequestSentError,
     OptionsTooMuchError,
     TakeoutInitDelayError,
     UserAlreadyParticipantError,
@@ -171,6 +172,8 @@ async def _join(
             return (await evt.sender.client(ImportChatInviteRequest(identifier))), None
         except UserAlreadyParticipantError:
             return None, await evt.reply("You are already in that chat.")
+        except InviteRequestSentError:
+            return None, await evt.reply("Invite request sent successfully.")
     else:
         channel = await evt.sender.client.get_entity(identifier)
         if not channel:

mautrix_telegram/config.py

@@ -84,6 +84,10 @@ class Config(BaseBridgeConfig):
 
         copy("appservice.provisioning.enabled")
         copy("appservice.provisioning.prefix")
+        if base["appservice.provisioning.prefix"].endswith("/v1"):
+            base["appservice.provisioning.prefix"] = base["appservice.provisioning.prefix"][
+                : -len("/v1")
+            ]
         copy("appservice.provisioning.shared_secret")
         if base["appservice.provisioning.shared_secret"] == "generate":
             base["appservice.provisioning.shared_secret"] = self._new_token()
@@ -137,7 +141,6 @@ class Config(BaseBridgeConfig):
         copy("bridge.inline_images")
         copy("bridge.image_as_file_size")
         copy("bridge.image_as_file_pixels")
-        copy("bridge.max_document_size")
         copy("bridge.parallel_file_transfer")
         copy("bridge.federate_rooms")
         copy("bridge.animated_sticker.target")
@@ -160,6 +163,7 @@ class Config(BaseBridgeConfig):
         copy("bridge.tag_only_on_create")
         copy("bridge.bridge_matrix_leave")
         copy("bridge.kick_on_logout")
+        copy("bridge.always_read_joined_telegram_notice")
         copy("bridge.backfill.invite_own_puppet")
         copy("bridge.backfill.takeout_limit")
         copy("bridge.backfill.initial_limit")

mautrix_telegram/db/message.py

@@ -21,7 +21,7 @@ from asyncpg import Record
 from attr import dataclass
 
 from mautrix.types import EventID, RoomID
-from mautrix.util.async_db import Database
+from mautrix.util.async_db import Database, Scheme
 
 from ..types import TelegramID
 
@@ -76,7 +76,7 @@ class Message:
     async def get_first_by_tgids(
         cls, tgids: list[TelegramID], tg_space: TelegramID
     ) -> list[Message]:
-        if cls.db.scheme == "postgres":
+        if cls.db.scheme in (Scheme.POSTGRES, Scheme.COCKROACH):
             q = (
                 f"SELECT {cls.columns} FROM message"
                 " WHERE tgid=ANY($1) AND tg_space=$2 AND edit_index=0"
@@ -123,7 +123,7 @@ class Message:
     async def get_by_mxids(
         cls, mxids: list[EventID], mx_room: RoomID, tg_space: TelegramID
     ) -> list[Message]:
-        if cls.db.scheme == "postgres":
+        if cls.db.scheme in (Scheme.POSTGRES, Scheme.COCKROACH):
             q = (
                 f"SELECT {cls.columns} FROM message"
                 " WHERE mxid=ANY($1) AND mx_room=$2 AND tg_space=$3"

mautrix_telegram/db/puppet.py

@@ -43,6 +43,7 @@ class Puppet:
     displayname_quality: int
     disable_updates: bool
     username: str | None
+    phone: str | None
     photo_id: str | None
     avatar_url: ContentURI | None
     name_set: bool
@@ -65,7 +66,7 @@ class Puppet:
 
     columns: ClassVar[str] = (
         "id, is_registered, displayname, displayname_source, displayname_contact, "
-        "displayname_quality, disable_updates, username, photo_id, avatar_url, "
+        "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"
     )
 
@@ -105,6 +106,7 @@ class Puppet:
             self.displayname_quality,
             self.disable_updates,
             self.username,
+            self.phone,
             self.photo_id,
             self.avatar_url,
             self.name_set,
@@ -121,9 +123,9 @@ class Puppet:
         q = """
         UPDATE puppet
         SET is_registered=$2, displayname=$3, displayname_source=$4, displayname_contact=$5,
-            displayname_quality=$6, disable_updates=$7, username=$8, photo_id=$9,
-            avatar_url=$10, name_set=$11, avatar_set=$12, is_bot=$13, is_channel=$14,
-            custom_mxid=$15, access_token=$16, next_batch=$17, base_url=$18
+            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
         WHERE id=$1
         """
         await self.db.execute(q, *self._values)
@@ -132,8 +134,9 @@ class Puppet:
         q = """
         INSERT INTO puppet (
             id, is_registered, displayname, displayname_source, displayname_contact,
-            displayname_quality, disable_updates, username, photo_id, avatar_url, name_set,
+            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
-        ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18)
+        ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,
+                  $19)
         """
         await self.db.execute(q, *self._values)

mautrix_telegram/db/telethon_session.py

@@ -24,7 +24,7 @@ from telethon.crypto import AuthKey
 from telethon.sessions import MemorySession
 from telethon.tl.types import PeerChannel, PeerChat, PeerUser, updates
 
-from mautrix.util.async_db import Database
+from mautrix.util.async_db import Database, Scheme
 
 fake_db = Database.create("") if TYPE_CHECKING else None
 
@@ -153,7 +153,7 @@ class PgSession(MemorySession):
         ] = self._entities_to_rows(tlo)
         if not rows:
             return
-        if self.db.scheme == "postgres":
+        if self.db.scheme == Scheme.POSTGRES:
             q = (
                 "INSERT INTO telethon_entities (session_id, id, hash, username, phone, name) "
                 "VALUES ($1, unnest($2::bigint[]), unnest($3::bigint[]), "
@@ -201,7 +201,7 @@ class PgSession(MemorySession):
             utils.get_peer_id(PeerChat(key)),
             utils.get_peer_id(PeerChannel(key)),
         )
-        if self.db.scheme == "postgres":
+        if self.db.scheme in (Scheme.POSTGRES, Scheme.COCKROACH):
             return await self._select_entity("id=ANY($1)", ids)
         else:
             return await self._select_entity(f"id IN ($1, $2, $3)", *ids)

mautrix_telegram/db/upgrade/__init__.py

@@ -9,4 +9,5 @@ from . import (
     v04_disappearing_messages,
     v05_channel_ghosts,
     v06_puppet_avatar_url,
+    v07_puppet_phone_number,
 )

mautrix_telegram/db/upgrade/v00_latest_revision.py

@@ -0,0 +1,200 @@
+# 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
+
+
+async def create_v7_tables(conn: Connection) -> int:
+    await conn.execute(
+        """CREATE TABLE "user" (
+            mxid TEXT   PRIMARY KEY,
+            tgid BIGINT UNIQUE,
+            tg_username    TEXT,
+            tg_phone       TEXT,
+            is_bot         BOOLEAN NOT NULL DEFAULT false,
+            saved_contacts INTEGER NOT NULL DEFAULT 0
+        )"""
+    )
+    await conn.execute(
+        """CREATE TABLE portal (
+            tgid        BIGINT,
+            tg_receiver BIGINT,
+            peer_type   TEXT NOT NULL,
+            mxid        TEXT UNIQUE,
+            avatar_url  TEXT,
+            encrypted   BOOLEAN NOT NULL DEFAULT false,
+            username    TEXT,
+            title       TEXT,
+            about       TEXT,
+            photo_id    TEXT,
+            name_set    BOOLEAN NOT NULL DEFAULT false,
+            avatar_set  BOOLEAN NOT NULL DEFAULT false,
+            megagroup   BOOLEAN,
+            config      jsonb,
+
+            sponsored_event_id     TEXT,
+            sponsored_event_ts     BIGINT,
+            sponsored_msg_random_id bytea,
+
+            PRIMARY KEY (tgid, tg_receiver)
+        )"""
+    )
+    await conn.execute(
+        """CREATE TABLE message (
+            mxid         TEXT   NOT NULL,
+            mx_room      TEXT   NOT NULL,
+            tgid         BIGINT,
+            tg_space     BIGINT,
+            edit_index   INTEGER,
+            redacted     BOOLEAN NOT NULL DEFAULT false,
+            content_hash bytea,
+            PRIMARY KEY (tgid, tg_space, edit_index),
+            UNIQUE (mxid, mx_room, tg_space)
+        )"""
+    )
+    await conn.execute(
+        """CREATE TABLE 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),
+            UNIQUE (mxid, mx_room)
+        )"""
+    )
+    await conn.execute(
+        """CREATE TABLE disappearing_message (
+            room_id             TEXT,
+            event_id            TEXT,
+            expiration_seconds  BIGINT,
+            expiration_ts       BIGINT,
+
+            PRIMARY KEY (room_id, event_id)
+        )"""
+    )
+    await conn.execute(
+        """CREATE TABLE puppet (
+            id BIGINT PRIMARY KEY,
+
+            is_registered BOOLEAN NOT NULL DEFAULT false,
+
+            displayname         TEXT,
+            displayname_source  BIGINT,
+            displayname_contact BOOLEAN NOT NULL DEFAULT true,
+            displayname_quality INTEGER NOT NULL DEFAULT 0,
+            disable_updates     BOOLEAN NOT NULL DEFAULT false,
+            username            TEXT,
+            phone               TEXT,
+            photo_id            TEXT,
+            avatar_url          TEXT,
+            name_set            BOOLEAN NOT NULL DEFAULT false,
+            avatar_set          BOOLEAN NOT NULL DEFAULT false,
+            is_bot              BOOLEAN,
+            is_channel          BOOLEAN NOT NULL DEFAULT false,
+
+            access_token TEXT,
+            custom_mxid  TEXT,
+            next_batch   TEXT,
+            base_url     TEXT
+        )"""
+    )
+    await conn.execute(
+        """CREATE TABLE telegram_file (
+            id              TEXT PRIMARY KEY,
+            mxc             TEXT NOT NULL,
+            mime_type       TEXT,
+            was_converted   BOOLEAN NOT NULL DEFAULT false,
+            timestamp       BIGINT  NOT NULL DEFAULT 0,
+            size            BIGINT,
+            width           INTEGER,
+            height          INTEGER,
+            thumbnail       TEXT,
+            decryption_info jsonb,
+            FOREIGN KEY (thumbnail) REFERENCES telegram_file(id)
+                ON UPDATE CASCADE ON DELETE SET NULL
+        )"""
+    )
+    await conn.execute(
+        """CREATE TABLE bot_chat (
+            id   BIGINT PRIMARY KEY,
+            type TEXT NOT NULL
+        )"""
+    )
+    await conn.execute(
+        """CREATE TABLE user_portal (
+            "user"          BIGINT,
+            portal          BIGINT,
+            portal_receiver BIGINT,
+            PRIMARY KEY ("user", portal, portal_receiver),
+            FOREIGN KEY ("user") REFERENCES "user"(tgid) ON DELETE CASCADE ON UPDATE CASCADE,
+            FOREIGN KEY (portal, portal_receiver) REFERENCES portal(tgid, tg_receiver)
+                 ON DELETE CASCADE ON UPDATE CASCADE
+        )"""
+    )
+    await conn.execute(
+        """CREATE TABLE contact (
+            "user"  BIGINT,
+            contact BIGINT,
+            PRIMARY KEY ("user", contact),
+            FOREIGN KEY ("user")  REFERENCES "user"(tgid) ON DELETE CASCADE ON UPDATE CASCADE,
+            FOREIGN KEY (contact) REFERENCES puppet(id)   ON DELETE CASCADE ON UPDATE CASCADE
+        )"""
+    )
+    await conn.execute(
+        """CREATE TABLE telethon_sessions (
+            session_id     TEXT PRIMARY KEY,
+            dc_id          INTEGER,
+            server_address TEXT,
+            port           INTEGER,
+            auth_key       bytea
+        )"""
+    )
+    await conn.execute(
+        """CREATE TABLE telethon_entities (
+            session_id TEXT,
+            id         BIGINT,
+            hash       BIGINT NOT NULL,
+            username   TEXT,
+            phone      TEXT,
+            name       TEXT,
+            PRIMARY KEY (session_id, id)
+        )"""
+    )
+    await conn.execute(
+        """CREATE TABLE telethon_sent_files (
+            session_id TEXT,
+            md5_digest bytea,
+            file_size  INTEGER,
+            type       INTEGER,
+            id         BIGINT,
+            hash       BIGINT,
+            PRIMARY KEY (session_id, md5_digest, file_size, type)
+        )"""
+    )
+    await conn.execute(
+        """CREATE TABLE telethon_update_state (
+            session_id   TEXT,
+            entity_id    BIGINT,
+            pts          BIGINT,
+            qts          BIGINT,
+            date         BIGINT,
+            seq          BIGINT,
+            unread_count INTEGER,
+            PRIMARY KEY (session_id, entity_id)
+        )"""
+    )
+    return 7

mautrix_telegram/db/upgrade/v01_initial_revision.py

@@ -15,29 +15,38 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 from __future__ import annotations
 
-from asyncpg import Connection
+from mautrix.util.async_db import Connection, Scheme
 
 from . import upgrade_table
+from .v00_latest_revision import create_v7_tables
 
 legacy_version_query = "SELECT version_num FROM alembic_version"
 last_legacy_version = "bfc0a39bfe02"
 
 
 def table_exists(scheme: str, name: str) -> str:
-    if scheme == "sqlite":
+    if scheme == Scheme.SQLITE:
         return f"SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='{name}')"
-    elif scheme == "postgres":
-        return f"SELECT EXISTS(SELECT FROM information_schema.tables WHERE table_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")
 
 
-@upgrade_table.register(description="Initial asyncpg revision")
-async def upgrade_v1(conn: Connection, scheme: str) -> None:
+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 v7.
+    # If it's a new db, we'll create the v7 tables directly (see the create_v7_tables call).
+    return 1 if is_legacy else 7
+
+
+@upgrade_table.register(description="Initial asyncpg revision", upgrades_to=first_upgrade_target)
+async def upgrade_v1(conn: Connection, scheme: str) -> int:
     is_legacy = await conn.fetchval(table_exists(scheme, "alembic_version"))
     if is_legacy:
         await migrate_legacy_to_v1(conn, scheme)
+        return 1
     else:
-        await create_v1_tables(conn)
+        return await create_v7_tables(conn)
 
 
 async def drop_constraints(conn: Connection, table: str, contype: str) -> None:
@@ -178,151 +187,3 @@ async def varchar_to_text(conn: Connection) -> None:
     for table, columns in columns_to_adjust.items():
         for column in columns:
             await conn.execute(f'ALTER TABLE "{table}" ALTER COLUMN {column} TYPE TEXT')
-
-
-async def create_v1_tables(conn: Connection) -> None:
-    await conn.execute(
-        """CREATE TABLE "user" (
-            mxid TEXT   PRIMARY KEY,
-            tgid BIGINT UNIQUE,
-            tg_username    TEXT,
-            tg_phone       TEXT,
-            is_bot         BOOLEAN NOT NULL DEFAULT false,
-            saved_contacts INTEGER NOT NULL DEFAULT 0
-        )"""
-    )
-    await conn.execute(
-        """CREATE TABLE portal (
-            tgid        BIGINT,
-            tg_receiver BIGINT,
-            peer_type   TEXT NOT NULL,
-            mxid        TEXT UNIQUE,
-            avatar_url  TEXT,
-            encrypted   BOOLEAN NOT NULL DEFAULT false,
-            username    TEXT,
-            title       TEXT,
-            about       TEXT,
-            photo_id    TEXT,
-            megagroup   BOOLEAN,
-            config      jsonb,
-            PRIMARY KEY (tgid, tg_receiver)
-        )"""
-    )
-    await conn.execute(
-        """CREATE TABLE message (
-            mxid       TEXT,
-            mx_room    TEXT,
-            tgid       BIGINT NOT NULL,
-            tg_space   BIGINT NOT NULL,
-            edit_index INTEGER NOT NULL,
-            redacted   BOOLEAN NOT NULL DEFAULT false,
-            PRIMARY KEY (tgid, tg_space, edit_index),
-            UNIQUE (mxid, mx_room, tg_space)
-        )"""
-    )
-    await conn.execute(
-        """CREATE TABLE puppet (
-            id BIGINT PRIMARY KEY,
-
-            is_registered BOOLEAN NOT NULL DEFAULT false,
-
-            displayname         TEXT,
-            displayname_source  BIGINT,
-            displayname_contact BOOLEAN NOT NULL DEFAULT true,
-            displayname_quality INTEGER NOT NULL DEFAULT 0,
-            disable_updates     BOOLEAN NOT NULL DEFAULT false,
-            username            TEXT,
-            photo_id            TEXT,
-            is_bot              BOOLEAN,
-
-            access_token TEXT,
-            custom_mxid  TEXT,
-            next_batch   TEXT,
-            base_url     TEXT
-        )"""
-    )
-    await conn.execute(
-        """CREATE TABLE telegram_file (
-            id              TEXT PRIMARY KEY,
-            mxc             TEXT NOT NULL,
-            mime_type       TEXT,
-            was_converted   BOOLEAN NOT NULL DEFAULT false,
-            timestamp       BIGINT  NOT NULL DEFAULT 0,
-            size            BIGINT,
-            width           INTEGER,
-            height          INTEGER,
-            thumbnail       TEXT,
-            decryption_info jsonb,
-            FOREIGN KEY (thumbnail) REFERENCES telegram_file(id)
-                ON UPDATE CASCADE ON DELETE SET NULL
-        )"""
-    )
-    await conn.execute(
-        """CREATE TABLE bot_chat (
-            id   BIGINT PRIMARY KEY,
-            type TEXT NOT NULL
-        )"""
-    )
-    await conn.execute(
-        """CREATE TABLE user_portal (
-            "user"          BIGINT,
-            portal          BIGINT,
-            portal_receiver BIGINT,
-            PRIMARY KEY ("user", portal, portal_receiver),
-            FOREIGN KEY ("user") REFERENCES "user"(tgid) ON DELETE CASCADE ON UPDATE CASCADE,
-            FOREIGN KEY (portal, portal_receiver) REFERENCES portal(tgid, tg_receiver)
-                 ON DELETE CASCADE ON UPDATE CASCADE
-        )"""
-    )
-    await conn.execute(
-        """CREATE TABLE contact (
-            "user"  BIGINT,
-            contact BIGINT,
-            PRIMARY KEY ("user", contact),
-            FOREIGN KEY ("user")  REFERENCES "user"(tgid) ON DELETE CASCADE ON UPDATE CASCADE,
-            FOREIGN KEY (contact) REFERENCES puppet(id)   ON DELETE CASCADE ON UPDATE CASCADE
-        )"""
-    )
-    await conn.execute(
-        """CREATE TABLE telethon_sessions (
-            session_id     TEXT PRIMARY KEY,
-            dc_id          INTEGER,
-            server_address TEXT,
-            port           INTEGER,
-            auth_key       bytea
-        )"""
-    )
-    await conn.execute(
-        """CREATE TABLE telethon_entities (
-            session_id TEXT,
-            id         BIGINT,
-            hash       BIGINT NOT NULL,
-            username   TEXT,
-            phone      TEXT,
-            name       TEXT,
-            PRIMARY KEY (session_id, id)
-        )"""
-    )
-    await conn.execute(
-        """CREATE TABLE telethon_sent_files (
-            session_id TEXT,
-            md5_digest bytea,
-            file_size  INTEGER,
-            type       INTEGER,
-            id         BIGINT,
-            hash       BIGINT,
-            PRIMARY KEY (session_id, md5_digest, file_size, type)
-        )"""
-    )
-    await conn.execute(
-        """CREATE TABLE telethon_update_state (
-            session_id   TEXT,
-            entity_id    BIGINT,
-            pts          BIGINT,
-            qts          BIGINT,
-            date         BIGINT,
-            seq          BIGINT,
-            unread_count INTEGER,
-            PRIMARY KEY (session_id, entity_id)
-        )"""
-    )

mautrix_telegram/db/upgrade/v02_sponsored_events.py

@@ -13,7 +13,7 @@
 #
 # 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 asyncpg import Connection
+from mautrix.util.async_db import Connection
 
 from . import upgrade_table
 

mautrix_telegram/db/upgrade/v03_reactions.py

@@ -13,7 +13,7 @@
 #
 # 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 asyncpg import Connection
+from mautrix.util.async_db import Connection
 
 from . import upgrade_table
 

mautrix_telegram/db/upgrade/v04_disappearing_messages.py

@@ -13,7 +13,7 @@
 #
 # 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 asyncpg import Connection
+from mautrix.util.async_db import Connection
 
 from . import upgrade_table
 

mautrix_telegram/db/upgrade/v05_channel_ghosts.py

@@ -13,7 +13,7 @@
 #
 # 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 asyncpg import Connection
+from mautrix.util.async_db import Connection, Scheme
 
 from . import upgrade_table
 
@@ -21,5 +21,5 @@ from . import upgrade_table
 @upgrade_table.register(description="Add separate ghost users for channel senders")
 async def upgrade_v5(conn: Connection, scheme: str) -> None:
     await conn.execute("ALTER TABLE puppet ADD COLUMN is_channel BOOLEAN NOT NULL DEFAULT false")
-    if scheme == "postgres":
+    if scheme == Scheme.POSTGRES:
         await conn.execute("ALTER TABLE puppet ALTER COLUMN is_channel DROP DEFAULT")

mautrix_telegram/db/upgrade/v06_puppet_avatar_url.py

@@ -13,7 +13,7 @@
 #
 # 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 asyncpg import Connection
+from mautrix.util.async_db import Connection
 
 from . import upgrade_table
 

mautrix_telegram/db/upgrade/v07_puppet_phone_number.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 phone number in puppet table")
+async def upgrade_v7(conn: Connection) -> None:
+    await conn.execute("ALTER TABLE puppet ADD COLUMN phone TEXT")

mautrix_telegram/db/user.py

@@ -21,7 +21,7 @@ from asyncpg import Record
 from attr import dataclass
 
 from mautrix.types import UserID
-from mautrix.util.async_db import Database
+from mautrix.util.async_db import Database, Scheme
 
 from ..types import TelegramID
 
@@ -104,7 +104,7 @@ class User:
         records = [(self.tgid, puppet_id) for puppet_id in puppets]
         async with self.db.acquire() as conn, conn.transaction():
             await conn.execute('DELETE FROM contact WHERE "user"=$1', self.tgid)
-            if self.db.scheme == "postgres":
+            if self.db.scheme == Scheme.POSTGRES:
                 await conn.copy_records_to_table("contact", records=records, columns=columns)
             else:
                 q = 'INSERT INTO contact ("user", contact) VALUES ($1, $2)'
@@ -120,7 +120,7 @@ class User:
         records = [(self.tgid, tgid, tg_receiver) for tgid, tg_receiver in portals]
         async with self.db.acquire() as conn, conn.transaction():
             await conn.execute('DELETE FROM user_portal WHERE "user"=$1', self.tgid)
-            if self.db.scheme == "postgres":
+            if self.db.scheme == Scheme.POSTGRES:
                 await conn.copy_records_to_table("user_portal", records=records, columns=columns)
             else:
                 q = 'INSERT INTO user_portal ("user", portal, portal_receiver) VALUES ($1, $2, $3)'

mautrix_telegram/example-config.yaml

@@ -16,6 +16,9 @@ homeserver:
     status_endpoint: null
     # Endpoint for reporting per-message status.
     message_send_checkpoint_endpoint: null
+    # Whether asynchronous uploads via MSC2246 should be enabled for media.
+    # Requires a media repo that supports MSC2246.
+    async_media: false
 
 # Application service host/registration related details
 # Changing these values requires regeneration of the registration.
@@ -64,7 +67,7 @@ appservice:
         # Whether or not the provisioning API should be enabled.
         enabled: true
         # The prefix to use in the provisioning API endpoints.
-        prefix: /_matrix/provision/v1
+        prefix: /_matrix/provision
         # The shared secret to authorize users of the API.
         # Set to "generate" to generate and save a new token.
         shared_secret: generate
@@ -210,8 +213,6 @@ bridge:
     image_as_file_size: 10
     # Maximum number of pixels in an image before sending to Telegram as a document. Defaults to 1280x1280 = 1638400.
     image_as_file_pixels: 1638400
-    # Maximum size of Telegram documents in megabytes to bridge.
-    max_document_size: 100
     # 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.
@@ -281,6 +282,8 @@ bridge:
     bridge_matrix_leave: true
     # Should the user be kicked out of all portals when logging out of the bridge?
     kick_on_logout: true
+    # Should the "* user joined Telegram" notice always be marked as read automatically?
+    always_read_joined_telegram_notice: true
     # Settings for backfilling messages from Telegram.
     backfill:
         # Whether or not the Telegram ghosts of logged in Matrix users should be

mautrix_telegram/formatter/from_telegram.py

@@ -20,7 +20,7 @@ import logging
 import re
 
 from telethon.errors import RPCError
-from telethon.helpers import add_surrogate, del_surrogate
+from telethon.helpers import add_surrogate, del_surrogate, within_surrogate
 from telethon.tl.custom import Message
 from telethon.tl.types import (
     MessageEntityBlockquote,
@@ -52,9 +52,9 @@ from mautrix.appservice import IntentAPI
 from mautrix.types import (
     EventType,
     Format,
+    InReplyTo,
     MessageType,
     RelatesTo,
-    RelationType,
     TextMessageEventContent,
 )
 
@@ -74,16 +74,13 @@ async def telegram_reply_to_matrix(evt: Message, source: au.AbstractUser) -> Rel
         )
         msg = await DBMessage.get_one_by_tgid(TelegramID(evt.reply_to.reply_to_msg_id), space)
         if msg:
-            return RelatesTo(rel_type=RelationType.REPLY, event_id=msg.mxid)
+            return RelatesTo(in_reply_to=InReplyTo(event_id=msg.mxid))
     return None
 
 
 async def _add_forward_header(
     source: au.AbstractUser, content: TextMessageEventContent, fwd_from: MessageFwdHeader
 ) -> None:
-    if not content.formatted_body or content.format != Format.HTML:
-        content.format = Format.HTML
-        content.formatted_body = escape(content.body)
     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))
@@ -139,6 +136,7 @@ async def _add_forward_header(
         fwd_from_text = "unknown source"
         fwd_from_html = f"unknown source"
 
+    content.ensure_has_html()
     content.body = "\n".join([f"> {line}" for line in content.body.split("\n")])
     content.body = f"Forwarded from {fwd_from_text}:\n{content.body}"
     content.formatted_body = (
@@ -160,8 +158,6 @@ async def _add_reply_header(
     if not msg:
         return
 
-    content.relates_to = RelatesTo(rel_type=RelationType.REPLY, event_id=msg.mxid)
-
     try:
         event = await main_intent.get_event(msg.mx_room, msg.mxid)
         if event.type == EventType.ROOM_ENCRYPTED and source.bridge.matrix.e2ee:
@@ -172,6 +168,7 @@ async def _add_reply_header(
         content.set_reply(event, displayname=puppet.displayname if puppet else event.sender)
     except Exception:
         log.exception("Failed to get event to add reply fallback")
+        content.set_reply(msg.mxid)
 
 
 async def telegram_to_matrix(
@@ -193,18 +190,13 @@ async def telegram_to_matrix(
     if entities:
         content.format = Format.HTML
         html = await _telegram_entities_to_matrix_catch(add_surrogate(content.body), entities)
-        content.formatted_body = del_surrogate(html).replace("\n", "<br/>")
-
-    def force_html():
-        if not content.formatted_body:
-            content.format = Format.HTML
-            content.formatted_body = escape(content.body).replace("\n", "<br/>")
+        content.formatted_body = del_surrogate(html)
 
     if require_html:
-        force_html()
+        content.ensure_has_html()
 
     if prefix_html:
-        force_html()
+        content.ensure_has_html()
         content.formatted_body = prefix_html + content.formatted_body
     if prefix_text:
         content.body = prefix_text + content.body
@@ -216,7 +208,7 @@ async def telegram_to_matrix(
         await _add_reply_header(source, content, evt, main_intent)
 
     if isinstance(evt, Message) and evt.post and evt.post_author:
-        force_html()
+        content.ensure_has_html()
         content.body += f"\n- {evt.post_author}"
         content.formatted_body += f"<br/><i>- <u>{evt.post_author}</u></i>"
 
@@ -234,30 +226,51 @@ async def _telegram_entities_to_matrix_catch(text: str, entities: list[TypeMessa
 
 
 async def _telegram_entities_to_matrix(
-    text: str, entities: list[TypeMessageEntity], offset: int = 0, length: int = None
+    text: str,
+    entities: list[TypeMessageEntity],
+    offset: int = 0,
+    length: int = None,
+    in_codeblock: bool = False,
 ) -> str:
+    def text_to_html(
+        val: str, _in_codeblock: bool = in_codeblock, escape_html: bool = True
+    ) -> str:
+        if escape_html:
+            val = escape(val)
+        if not _in_codeblock:
+            val = val.replace("\n", "<br/>")
+        return val
+
     if not entities:
-        return escape(text)
+        return text_to_html(text)
     if length is None:
         length = len(text)
     html = []
     last_offset = 0
     for i, entity in enumerate(entities):
-        if entity.offset > offset + length:
+        if entity.offset >= offset + length:
             break
         relative_offset = entity.offset - offset
         if relative_offset > last_offset:
-            html.append(escape(text[last_offset:relative_offset]))
+            html.append(text_to_html(text[last_offset:relative_offset]))
         elif relative_offset < last_offset:
             continue
 
+        while within_surrogate(text, relative_offset, length=length):
+            relative_offset += 1
+        while within_surrogate(text, relative_offset + entity.length, length=length):
+            entity.length += 1
+
         skip_entity = False
+        is_code_entity = isinstance(entity, (MessageEntityCode, MessageEntityPre))
         entity_text = await _telegram_entities_to_matrix(
             text=text[relative_offset : relative_offset + entity.length],
             entities=entities[i + 1 :],
             offset=entity.offset,
             length=entity.length,
+            in_codeblock=is_code_entity,
         )
+        entity_text = text_to_html(entity_text, is_code_entity, escape_html=False)
         entity_type = type(entity)
 
         if entity_type == MessageEntityBold:
@@ -285,7 +298,7 @@ async def _telegram_entities_to_matrix(
         elif entity_type == MessageEntityEmail:
             html.append(f"<a href='mailto:{entity_text}'>{entity_text}</a>")
         elif entity_type in (MessageEntityTextUrl, MessageEntityUrl):
-            skip_entity = await _parse_url(
+            await _parse_url(
                 html, entity_text, entity.url if entity_type == MessageEntityTextUrl else None
             )
         elif entity_type in (
@@ -300,7 +313,7 @@ async def _telegram_entities_to_matrix(
         else:
             skip_entity = True
         last_offset = relative_offset + (0 if skip_entity else entity.length)
-    html.append(escape(text[last_offset:]))
+    html.append(text_to_html(text[last_offset:]))
 
     return "".join(html)
 
@@ -316,12 +329,24 @@ def _parse_pre(html: list[str], entity_text: str, language: str) -> bool:
 async def _parse_mention(html: list[str], entity_text: str) -> bool:
     username = entity_text[1:]
 
+    mxid = None
+    portal = None
+    # This is a bit complicated because public channels have both Puppet and Portal instances.
+    # Basically the currently intended output is:
+    # User/bot mention (bridge user)          -> real user mention
+    # User/bot mention (normal Telegram user) -> ghost user mention
+    # Public channel with existing portal     -> room mention
+    # Public channel without portal           -> ghost user mention
+    # Other chat                              -> room mention
     user = await u.User.find_by_username(username) or await pu.Puppet.find_by_username(username)
     if user:
+        if isinstance(user, pu.Puppet) and user.is_channel:
+            portal = await po.Portal.get_by_tgid(user.tgid)
         mxid = user.mxid
     else:
         portal = await po.Portal.find_by_username(username)
-        mxid = portal.alias or portal.mxid if portal else None
+    if portal and (portal.mxid or not user):
+        mxid = portal.alias or portal.mxid
 
     if mxid:
         html.append(f"<a href='https://matrix.to/#/{mxid}'>{entity_text}</a>")
@@ -345,11 +370,15 @@ async def _parse_name_mention(html: list[str], entity_text: str, user_id: Telegr
 
 
 message_link_regex = re.compile(
-    r"https?://t(?:elegram)?\.(?:me|dog)/([A-Za-z][A-Za-z0-9_]{3,}[A-Za-z0-9])/([0-9]{1,50})"
+    r"https?://t(?:elegram)?\.(?:me|dog)"
+    # /username or /c/id
+    r"/([A-Za-z][A-Za-z0-9_]{3,31}[A-Za-z0-9]|[Cc]/[0-9]{1,20})"
+    # /messageid
+    r"/([0-9]{1,20})"
 )
 
 
-async def _parse_url(html: list[str], entity_text: str, url: str) -> bool:
+async def _parse_url(html: list[str], entity_text: str, url: str):
     url = escape(url) if url else entity_text
     if not url.startswith(("https://", "http://", "ftp://", "magnet://")):
         url = "http://" + url
@@ -359,11 +388,13 @@ async def _parse_url(html: list[str], entity_text: str, url: str) -> bool:
         group, msgid_str = message_link_match.groups()
         msgid = int(msgid_str)
 
-        portal = await po.Portal.find_by_username(group)
+        if group.lower().startswith("c/"):
+            portal = await po.Portal.get_by_tgid(TelegramID(int(group[2:])))
+        else:
+            portal = await po.Portal.find_by_username(group)
         if portal:
             message = await DBMessage.get_one_by_tgid(TelegramID(msgid), portal.tgid)
             if message:
                 url = f"https://matrix.to/#/{portal.mxid}/{message.mxid}"
 
     html.append(f"<a href='{url}'>{entity_text}</a>")
-    return False

mautrix_telegram/matrix.py

@@ -61,98 +61,22 @@ class MatrixHandler(BaseMatrixHandler):
 
         self._previously_typing = {}
 
-    async def handle_puppet_invite(
-        self, room_id: RoomID, puppet: pu.Puppet, inviter: u.User, event_id: EventID
+    async def handle_puppet_group_invite(
+        self,
+        room_id: RoomID,
+        puppet: pu.Puppet,
+        invited_by: u.User,
+        evt: StateEvent,
+        members: list[UserID],
     ) -> None:
-        intent = puppet.default_mxid_intent
-        self.log.debug(f"{inviter.mxid} invited puppet for {puppet.tgid} to {room_id}")
-        if puppet.is_channel:
-            self.log.debug(f"Rejecting invite for {puppet.tgid} to {room_id}: puppet is a channel")
-            await intent.leave_room(room_id, reason="Channels can't be invited to chats")
-            return
-
-        if not await inviter.is_logged_in():
-            self.log.debug(f"Rejecting invite for {puppet.tgid} to {room_id}: user not logged in")
-            await intent.leave_room(
-                room_id,
-                reason="Only users who are logged into the bridge can invite Telegram ghosts.",
-            )
-            return
-
-        portal = await po.Portal.get_by_mxid(room_id)
-        if portal:
-            if portal.peer_type == "user":
-                await intent.error_and_leave(
-                    room_id, text="You can not invite additional users to private chats."
-                )
-                return
-            await portal.invite_telegram(inviter, puppet)
-            await intent.join_room(room_id)
-            return
-        try:
-            members = await intent.get_room_members(room_id)
-        except MatrixError:
-            self.log.exception(f"Failed to get members after joining {room_id} as {intent.mxid}")
-            return
         if self.az.bot_mxid not in members:
-            if len(members) > 2:
-                await intent.error_and_leave(
-                    room_id,
-                    text=None,
-                    html=(
-                        f"Please invite "
-                        f"<a href='https://matrix.to/#/{self.az.bot_mxid}'>the bridge bot</a> "
-                        f"first if you want to create a Telegram chat."
-                    ),
-                )
-                return
-
-            await intent.join_room(room_id)
-            portal = await po.Portal.get_by_tgid(
-                puppet.tgid, tg_receiver=inviter.tgid, peer_type="user"
+            await puppet.default_mxid_intent.leave_room(
+                room_id, reason="This ghost does not join multi-user rooms without the bridge bot."
             )
-            if portal.mxid:
-                try:
-                    await portal.invite_to_matrix(inviter.mxid)
-                    await intent.send_notice(
-                        room_id,
-                        text=f"You already have a private chat with me: {portal.mxid}",
-                        html=(
-                            "You already have a private chat with me: "
-                            f"<a href='https://matrix.to/#/{portal.mxid}'>Link to room</a>"
-                        ),
-                    )
-                    await intent.leave_room(room_id)
-                    return
-                except MatrixError:
-                    pass
-            portal.mxid = room_id
-            e2be_ok = await portal.check_dm_encryption()
-            await portal.save()
-            await inviter.register_portal(portal)
-            if e2be_ok is True:
-                evt_type, content = await self.e2ee.encrypt(
-                    room_id,
-                    EventType.ROOM_MESSAGE,
-                    TextMessageEventContent(
-                        msgtype=MessageType.NOTICE,
-                        body=(
-                            "Portal to private chat created and end-to-bridge encryption enabled."
-                        ),
-                    ),
-                )
-                await intent.send_message_event(room_id, evt_type, content)
-            else:
-                message = "Portal to private chat created."
-                if e2be_ok is False:
-                    message += "\n\nWarning: Failed to enable end-to-bridge encryption"
-                await intent.send_notice(room_id, message)
-            await portal.update_bridge_info()
         else:
-            await intent.join_room(room_id)
-            await intent.send_notice(
+            await puppet.default_mxid_intent.send_notice(
                 room_id,
-                "This puppet will remain inactive until a Telegram chat is created for this room.",
+                "This ghost will remain inactive until a Telegram chat is created for this room.",
             )
 
     async def handle_invite(
@@ -163,9 +87,13 @@ class MatrixHandler(BaseMatrixHandler):
             return
         await user.ensure_started()
         portal = await po.Portal.get_by_mxid(room_id)
-        if user and await user.has_full_access(allow_bot=True):
-            if portal and portal.allow_bridging:
-                await portal.invite_telegram(inviter, user)
+        if (
+            user
+            and portal
+            and await user.has_full_access(allow_bot=True)
+            and portal.allow_bridging
+        ):
+            await portal.handle_matrix_invite(inviter, user)
 
     async def handle_join(self, room_id: RoomID, user_id: UserID, event_id: EventID) -> None:
         user = await u.User.get_and_start_by_mxid(user_id)

mautrix_telegram/portal.py

@@ -105,6 +105,7 @@ from telethon.tl.types import (
     MessageActionChatEditPhoto,
     MessageActionChatEditTitle,
     MessageActionChatJoinedByLink,
+    MessageActionChatJoinedByRequest,
     MessageActionChatMigrateTo,
     MessageActionContactSignUp,
     MessageActionGameScore,
@@ -163,7 +164,7 @@ from telethon.utils import decode_waveform
 import magic
 
 from mautrix.appservice import DOUBLE_PUPPET_SOURCE_KEY, IntentAPI
-from mautrix.bridge import BasePortal, NotificationDisabler, async_getter_lock
+from mautrix.bridge import BasePortal, NotificationDisabler, RejectMatrixInvite, async_getter_lock
 from mautrix.errors import IntentError, MatrixRequestError, MForbidden
 from mautrix.types import (
     ContentURI,
@@ -448,6 +449,14 @@ class Portal(DBPortal, BasePortal):
     # endregion
     # region Matrix -> Telegram metadata
 
+    async def save(self) -> None:
+        if self.deleted:
+            await super().insert()
+            await self.postinit()
+            self.deleted = False
+        else:
+            await super().save()
+
     async def get_telegram_users_in_matrix_room(
         self, source: u.User, pre_create: bool = False
     ) -> tuple[list[InputPeerUser], list[UserID]]:
@@ -570,18 +579,25 @@ class Portal(DBPortal, BasePortal):
         await self.handle_matrix_power_levels(source, levels.users, {}, None)
         await self.update_bridge_info()
 
-    async def invite_telegram(self, source: u.User, puppet: p.Puppet | au.AbstractUser) -> None:
+    async def handle_matrix_invite(
+        self, invited_by: u.User, puppet: p.Puppet | au.AbstractUser
+    ) -> None:
         if puppet.is_channel:
             raise ValueError("Can't invite channels to chats")
-        if self.peer_type == "chat":
-            await source.client(
-                AddChatUserRequest(chat_id=self.tgid, user_id=puppet.tgid, fwd_limit=0)
-            )
-        elif self.peer_type == "channel":
-            await source.client(InviteToChannelRequest(channel=self.peer, users=[puppet.tgid]))
-        # We don't care if there are invites for private chat portals with the relaybot.
-        elif not self.bot or self.tg_receiver != self.bot.tgid:
-            raise ValueError("Invalid peer type for Telegram user invite")
+        try:
+            if self.peer_type == "chat":
+                await invited_by.client(
+                    AddChatUserRequest(chat_id=self.tgid, user_id=puppet.tgid, fwd_limit=0)
+                )
+            elif self.peer_type == "channel":
+                await invited_by.client(
+                    InviteToChannelRequest(channel=self.peer, users=[puppet.tgid])
+                )
+            # We don't care if there are invites for private chat portals with the relaybot.
+            elif not self.bot or self.tg_receiver != self.bot.tgid:
+                raise RejectMatrixInvite("You can't invite additional users to private chats.")
+        except RPCError as e:
+            raise RejectMatrixInvite(e.message) from e
 
     # endregion
     # region Telegram -> Matrix metadata
@@ -613,15 +629,12 @@ class Portal(DBPortal, BasePortal):
         self,
         user: au.AbstractUser,
         entity: TypeChat | User,
-        direct: bool = None,
         puppet: p.Puppet = None,
         levels: PowerLevelStateEventContent = None,
         users: list[User] = None,
     ) -> None:
-        if direct is None:
-            direct = self.peer_type == "user"
         try:
-            await self._update_matrix_room(user, entity, direct, puppet, levels, users)
+            await self._update_matrix_room(user, entity, puppet, levels, users)
         except Exception:
             self.log.exception("Fatal error updating Matrix room")
 
@@ -629,12 +642,11 @@ class Portal(DBPortal, BasePortal):
         self,
         user: au.AbstractUser,
         entity: TypeChat | User,
-        direct: bool,
         puppet: p.Puppet = None,
         levels: PowerLevelStateEventContent = None,
         users: list[User] = None,
     ) -> None:
-        if not direct:
+        if not self.is_direct:
             await self.update_info(user, entity)
             if not users:
                 users = await self._get_users(user, entity)
@@ -642,7 +654,7 @@ class Portal(DBPortal, BasePortal):
             await self.update_power_levels(users, levels)
         else:
             if not puppet:
-                puppet = await p.Puppet.get_by_tgid(self.tgid)
+                puppet = await self.get_dm_puppet()
             await puppet.update_info(user, entity)
             await puppet.intent_for(self).join_room(self.mxid)
             await self.update_info_from_puppet(puppet, user, entity.photo)
@@ -661,12 +673,14 @@ class Portal(DBPortal, BasePortal):
 
     async def update_info_from_puppet(
         self,
-        puppet: p.Puppet,
+        puppet: p.Puppet | None = None,
         source: au.AbstractUser | None = None,
         photo: UserProfilePhoto | None = None,
     ) -> None:
         if not self.encrypted and not self.private_chat_portal_meta:
             return
+        if puppet is None:
+            puppet = await self.get_dm_puppet()
         # The bridge bot needs to join for e2ee, but that messes up the default name
         # generation. If/when canonical DMs happen, this might not be necessary anymore.
         changed = await self._update_avatar_from_puppet(puppet, source, photo)
@@ -690,7 +704,7 @@ class Portal(DBPortal, BasePortal):
                     except Exception:
                         self.log.exception(f"Failed to get entity through {user.tgid} for update")
                         return self.mxid
-                update = self.update_matrix_room(user, entity, self.peer_type == "user")
+                update = self.update_matrix_room(user, entity)
                 asyncio.create_task(update)
                 await self.invite_to_matrix(invites or [])
             return self.mxid
@@ -754,7 +768,6 @@ class Portal(DBPortal, BasePortal):
         elif not self.allow_bridging:
             return None
 
-        direct = self.peer_type == "user"
         invites = invites or []
 
         if not entity:
@@ -768,14 +781,14 @@ class Portal(DBPortal, BasePortal):
         except AttributeError:
             self.title = None
 
-        if direct and self.tgid == user.tgid:
+        if self.is_direct and self.tgid == user.tgid:
             self.title = "Telegram Saved Messages"
             self.about = "Your Telegram cloud storage chat"
 
-        puppet = await p.Puppet.get_by_tgid(self.tgid) if direct else None
+        puppet = await self.get_dm_puppet()
         if puppet:
             await puppet.update_info(user, entity)
-        self._main_intent = puppet.intent_for(self) if direct else self.az.intent
+        self._main_intent = puppet.intent_for(self) if self.is_direct else self.az.intent
 
         if self.peer_type == "channel":
             self.megagroup = entity.megagroup
@@ -796,7 +809,7 @@ class Portal(DBPortal, BasePortal):
 
         power_levels = putil.get_base_power_levels(self, entity=entity)
         users = None
-        if not direct:
+        if not self.is_direct:
             users = await self._get_users(user, entity)
             if self.has_bot:
                 extra_invites = self.config["bridge.relaybot.group_chat_invite"]
@@ -836,9 +849,9 @@ class Portal(DBPortal, BasePortal):
                     "content": {"algorithm": "m.megolm.v1.aes-sha2"},
                 }
             )
-            if direct:
+            if self.is_direct:
                 create_invites.append(self.az.bot_mxid)
-        if direct and (self.encrypted or self.private_chat_portal_meta):
+        if self.is_direct and (self.encrypted or self.private_chat_portal_meta):
             self.title = puppet.displayname
             self.avatar_url = puppet.avatar_url
             self.photo_id = puppet.photo_id
@@ -857,7 +870,7 @@ class Portal(DBPortal, BasePortal):
             room_id = await self.main_intent.create_room(
                 alias_localpart=alias,
                 preset=preset,
-                is_direct=direct,
+                is_direct=self.is_direct,
                 invitees=create_invites,
                 name=self.title,
                 topic=self.about,
@@ -869,7 +882,7 @@ class Portal(DBPortal, BasePortal):
             self.name_set = bool(self.title)
             self.avatar_set = bool(self.avatar_url)
 
-            if self.encrypted and self.matrix.e2ee and direct:
+            if self.encrypted and self.matrix.e2ee and self.is_direct:
                 try:
                     await self.az.intent.ensure_joined(room_id)
                 except Exception:
@@ -884,9 +897,7 @@ class Portal(DBPortal, BasePortal):
             await self.invite_to_matrix(invites)
 
             update_room = asyncio.create_task(
-                self.update_matrix_room(
-                    user, entity, direct, puppet, levels=power_levels, users=users
-                )
+                self.update_matrix_room(user, entity, puppet, levels=power_levels, users=users)
             )
 
             if self.config["bridge.backfill.initial_limit"] > 0:
@@ -1118,9 +1129,10 @@ class Portal(DBPortal, BasePortal):
             return False
 
         self.about = about
-        await self._try_set_state(
-            sender, EventType.ROOM_TOPIC, RoomTopicStateEventContent(topic=self.about)
-        )
+        if self.mxid:
+            await self._try_set_state(
+                sender, EventType.ROOM_TOPIC, RoomTopicStateEventContent(topic=self.about)
+            )
         if save:
             await self.save()
         return True
@@ -1132,14 +1144,15 @@ class Portal(DBPortal, BasePortal):
             return False
 
         self.title = title
-        try:
-            await self._try_set_state(
-                sender, EventType.ROOM_NAME, RoomNameStateEventContent(name=self.title)
-            )
-            self.name_set = True
-        except Exception as e:
-            self.log.warning(f"Failed to set room name: {e}")
-            self.name_set = False
+        if self.mxid:
+            try:
+                await self._try_set_state(
+                    sender, EventType.ROOM_NAME, RoomNameStateEventContent(name=self.title)
+                )
+                self.name_set = True
+            except Exception as e:
+                self.log.warning(f"Failed to set room name: {e}")
+                self.name_set = False
         if save:
             await self.save()
         return True
@@ -1152,14 +1165,17 @@ class Portal(DBPortal, BasePortal):
         if puppet.avatar_url:
             self.photo_id = puppet.photo_id
             self.avatar_url = puppet.avatar_url
-            try:
-                await self._try_set_state(
-                    None, EventType.ROOM_AVATAR, RoomAvatarStateEventContent(url=self.avatar_url)
-                )
-                self.avatar_set = True
-            except Exception as e:
-                self.log.warning(f"Failed to set room avatar: {e}")
-                self.avatar_set = False
+            if self.mxid:
+                try:
+                    await self._try_set_state(
+                        None,
+                        EventType.ROOM_AVATAR,
+                        RoomAvatarStateEventContent(url=self.avatar_url),
+                    )
+                    self.avatar_set = True
+                except Exception as e:
+                    self.log.warning(f"Failed to set room avatar: {e}")
+                    self.avatar_set = False
             return True
         elif photo is not None and user is not None:
             return await self._update_avatar(user, photo=photo)
@@ -1197,19 +1213,27 @@ class Portal(DBPortal, BasePortal):
                 self.photo_id = ""
                 self.avatar_url = None
             elif self.photo_id != photo_id or not self.avatar_url:
-                file = await util.transfer_file_to_matrix(user.client, self.main_intent, loc)
+                file = await util.transfer_file_to_matrix(
+                    user.client,
+                    self.main_intent,
+                    loc,
+                    async_upload=self.config["homeserver.async_media"],
+                )
                 if not file:
                     return False
                 self.photo_id = photo_id
                 self.avatar_url = file.mxc
-            try:
-                await self._try_set_state(
-                    sender, EventType.ROOM_AVATAR, RoomAvatarStateEventContent(url=self.avatar_url)
-                )
-                self.avatar_set = True
-            except Exception as e:
-                self.log.warning(f"Failed to set room avatar: {e}")
-                self.avatar_set = False
+            if self.mxid:
+                try:
+                    await self._try_set_state(
+                        sender,
+                        EventType.ROOM_AVATAR,
+                        RoomAvatarStateEventContent(url=self.avatar_url),
+                    )
+                    self.avatar_set = True
+                except Exception as e:
+                    self.log.warning(f"Failed to set room avatar: {e}")
+                    self.avatar_set = False
             if save:
                 await self.save()
             return True
@@ -1401,7 +1425,7 @@ class Portal(DBPortal, BasePortal):
                 f"{message.mxid}/{message.tgid} as read by {user.mxid}/{user.tgid}"
             )
         await user.client.send_read_acknowledge(
-            self.peer, max_id=message.tgid, clear_mentions=True
+            self.peer, max_id=message.tgid, clear_mentions=True, clear_reactions=True
         )
         if self.peer_type == "channel" and not self.megagroup:
             asyncio.create_task(self._try_handle_read_for_sponsored_msg(user, event_id, timestamp))
@@ -1448,7 +1472,7 @@ class Portal(DBPortal, BasePortal):
                 del self.by_mxid[self.mxid]
             except KeyError:
                 pass
-        elif self.config["bridge.kick_on_logout"]:
+        elif self.config["bridge.bridge_matrix_leave"]:
             await user.client.delete_dialog(self.peer)
 
     async def join_matrix(self, user: u.User, event_id: EventID) -> None:
@@ -1485,7 +1509,9 @@ class Portal(DBPortal, BasePortal):
         return ruds[self.hash_user_id(user_id) % len(ruds)] if ruds else ""
 
     async def _apply_msg_format(self, sender: u.User, content: MessageEventContent) -> None:
-        if not isinstance(content, TextMessageEventContent) or content.format != Format.HTML:
+        if isinstance(content, TextMessageEventContent):
+            content.ensure_has_html()
+        else:
             content.format = Format.HTML
             content.formatted_body = escape_html(content.body).replace("\n", "<br/>")
 
@@ -1506,9 +1532,7 @@ class Portal(DBPortal, BasePortal):
         content.formatted_body = Template(tpl).safe_substitute(tpl_args)
 
     async def _apply_emote_format(self, sender: u.User, content: TextMessageEventContent) -> None:
-        if content.format != Format.HTML:
-            content.format = Format.HTML
-            content.formatted_body = escape_html(content.body).replace("\n", "<br/>")
+        content.ensure_has_html()
 
         tpl = self.get_config("emote_format")
         puppet = await p.Puppet.get_by_tgid(sender.tgid)
@@ -1634,6 +1658,9 @@ class Portal(DBPortal, BasePortal):
             attributes.append(DocumentAttributeImageSize(w, h))
             force_document = force_document or w * h >= max_image_pixels
 
+        if "fi.mau.telegram.force_document" in content:
+            force_document = bool(content["fi.mau.telegram.force_document"])
+
         if (mime == "image/png" or mime == "image/jpeg") and not force_document:
             media = InputMediaUploadedPhoto(file_handle)
         else:
@@ -2154,7 +2181,7 @@ class Portal(DBPortal, BasePortal):
                 "Failed to fully migrate to upgraded Matrix room: no Telegram user found."
             )
             return
-        await self.update_matrix_room(user, entity, direct=self.peer_type == "user")
+        await self.update_matrix_room(user, entity)
         self.log.info(f"{sender} upgraded room from {old_room} to {self.mxid}")
         await self._send_delivery_receipt(event_id, room_id=old_room)
 
@@ -2167,13 +2194,6 @@ class Portal(DBPortal, BasePortal):
         self.by_mxid[self.mxid] = self
         await self.save()
 
-    async def enable_dm_encryption(self) -> bool:
-        ok = await super().enable_dm_encryption()
-        if ok:
-            puppet = await p.Puppet.get_by_tgid(self.tgid)
-            await self.update_info_from_puppet(puppet)
-        return ok
-
     # endregion
     # region Telegram -> Matrix bridging
 
@@ -2210,7 +2230,11 @@ class Portal(DBPortal, BasePortal):
             )
             return await self._send_message(intent, content, timestamp=evt.date)
         file = await util.transfer_file_to_matrix(
-            source.client, intent, loc, encrypt=self.encrypted
+            source.client,
+            intent,
+            loc,
+            encrypt=self.encrypted,
+            async_upload=self.config["homeserver.async_media"],
         )
         if not file:
             return None
@@ -2354,7 +2378,7 @@ class Portal(DBPortal, BasePortal):
 
         attrs = self._parse_telegram_document_attributes(document.attributes)
 
-        if document.size > self.config["bridge.max_document_size"] * 1000**2:
+        if document.size > self.matrix.media_config.upload_size:
             name = attrs.name or ""
             caption = f"\n{evt.message}" if evt.message else ""
             # TODO encrypt
@@ -2376,6 +2400,7 @@ class Portal(DBPortal, BasePortal):
             filename=attrs.name,
             parallel_id=parallel_id,
             encrypt=self.encrypted,
+            async_upload=self.config["homeserver.async_media"],
         )
         if not file:
             return None
@@ -2423,6 +2448,8 @@ class Portal(DBPortal, BasePortal):
                 "image/": MessageType.IMAGE,
             }.get(info.mimetype[:6], MessageType.FILE),
         )
+        if event_type == EventType.STICKER:
+            content.msgtype = None
         if attrs.is_audio:
             content["org.matrix.msc1767.audio"] = {"duration": attrs.duration * 1000}
             if attrs.waveform:
@@ -2511,7 +2538,11 @@ class Portal(DBPortal, BasePortal):
             beeper_link_preview["og:image:height"] = largest_size.h
             beeper_link_preview["og:image:width"] = largest_size.w
             file = await util.transfer_file_to_matrix(
-                source.client, intent, loc, encrypt=self.encrypted
+                source.client,
+                intent,
+                loc,
+                encrypt=self.encrypted,
+                async_upload=self.config["homeserver.async_media"],
             )
 
             if file.decryption_info:
@@ -2758,7 +2789,7 @@ class Portal(DBPortal, BasePortal):
             "chats": self.peer_type == "chat",
             "users": self.peer_type == "user",
             "channels": (self.peer_type == "channel" and not self.megagroup),
-            "max_file_size": min(self.config["bridge.max_document_size"], 2000) * 1024 * 1024,
+            "max_file_size": min(self.matrix.media_config.upload_size, 2000 * 1024 * 1024),
         }
 
     async def backfill(
@@ -2776,7 +2807,7 @@ class Portal(DBPortal, BasePortal):
         source: u.User,
         is_initial: bool = False,
         limit: int | None = None,
-        last_id: int | None = None,
+        last_tgid: int | None = None,
     ) -> None:
         limit = limit or (
             self.config["bridge.backfill.initial_limit"]
@@ -2787,43 +2818,40 @@ class Portal(DBPortal, BasePortal):
             return
         if not self.config["bridge.backfill.normal_groups"] and self.peer_type == "chat":
             return
-        last = await DBMessage.find_last(
+        last_in_room = await DBMessage.find_last(
             self.mxid, (source.tgid if self.peer_type != "channel" else self.tgid)
         )
-        min_id = last.tgid if last else 0
-        if last_id is None:
+        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:
                 # The chat seems empty
                 return
-            last_id = messages[0].id
-        if last_id <= min_id:
+            last_tgid = messages[0].id
+        if last_tgid <= min_id or (last_tgid == 1 and self.peer_type == "channel"):
             # Nothing to backfill
             return
         if limit < 0:
-            limit = last_id - min_id
-            self.log.debug(
-                f"Backfilling approximately {last_id - min_id} messages through {source.mxid}"
-            )
+            limit = last_tgid - min_id
+            limit_type = "unlimited"
         elif self.peer_type == "channel":
-            # This is a channel or supergroup, so we'll backfill messages based on the ID.
-            # There are some cases, such as deleted messages, where this may backfill less
-            # messages than the limit.
-            min_id = max(last_id - limit, min_id)
-            self.log.debug(
-                f"Backfilling messages after ID {min_id} (last message: {last_id}) "
-                f"through {source.mxid}"
-            )
+            min_id = max(last_tgid - limit, min_id)
+            # This is now just an approximate message count, not the actual limit.
+            limit = last_tgid - min_id
+            limit_type = "channel"
         else:
-            # Private chats and normal groups don't have their own message ID namespace,
-            # which means we'll have to fetch messages a different way.
-            # The _backfill_messages method will detect min_id=None and not use reverse=True
-            min_id = None
-            self.log.debug(f"Backfilling up to {limit} messages through {source.mxid}")
+            # This limit will be higher than the actual message count if there are any messages
+            # in other DMs or normal groups, but that's not too bad.
+            limit = min(last_tgid - min_id, limit)
+            limit_type = "dm/minigroup"
+        self.log.debug(
+            f"Backfilling up to {limit} messages after ID {min_id} through {source.mxid} "
+            f"(last message: {last_tgid}, limit type: {limit_type})"
+        )
         with self.backfill_lock:
             await self._backfill(source, min_id, limit)
 
-    async def _backfill(self, source: u.User, min_id: int | None, limit: int) -> None:
+    async def _backfill(self, source: u.User, min_id: int, limit: int) -> None:
         self.backfill_leave = set()
         if (
             self.peer_type == "user"
@@ -2841,38 +2869,55 @@ class Portal(DBPortal, BasePortal):
             if limit > self.config["bridge.backfill.takeout_limit"]:
                 self.log.debug(f"Opening takeout client for {source.tgid}")
                 async with client.takeout(**self._takeout_options) as takeout:
-                    count = await self._backfill_messages(source, min_id, limit, takeout)
+                    count, handled = await self._backfill_messages(source, min_id, limit, takeout)
             else:
-                count = await self._backfill_messages(source, min_id, limit, client)
+                count, handled = await self._backfill_messages(source, min_id, limit, client)
 
         for intent in self.backfill_leave:
             self.log.trace("Leaving room with %s post-backfill", intent.mxid)
             await intent.leave_room(self.mxid)
         self.backfill_leave = None
-        self.log.info("Backfilled %d messages through %s", count, source.mxid)
+        self.log.info(
+            "Backfilled %d (of %d fetched) messages through %s", handled, count, source.mxid
+        )
 
     async def _backfill_messages(
-        self, source: u.User, min_id: int | None, limit: int, client: MautrixTelegramClient
-    ) -> int:
-        count = 0
+        self, source: u.User, min_id: int, limit: int, client: MautrixTelegramClient
+    ) -> tuple[int, int]:
+        count = handled_count = 0
         entity = await self.get_input_entity(source)
-        if min_id is not None:
+        if self.peer_type == "channel":
+            # This is a channel or supergroup, so we'll backfill messages based on the ID.
+            # There are some cases, such as deleted messages, where this may backfill less
+            # messages than the limit.
             self.log.debug(f"Iterating all messages starting with {min_id} (approx: {limit})")
             messages = client.iter_messages(entity, reverse=True, min_id=min_id)
             async for message in messages:
-                await self._handle_telegram_backfill_message(source, message)
                 count += 1
+                was_handled = await self._handle_telegram_backfill_message(source, message)
+                handled_count += 1 if was_handled else 0
         else:
-            self.log.debug(f"Fetching up to {limit} most recent messages")
-            messages = await client.get_messages(entity, limit=limit)
+            # Private chats and normal groups don't have their own message ID namespace,
+            # which means we'll have to fetch messages a different way.
+            self.log.debug(
+                f"Fetching up to {limit} most recent messages, ignoring anything before {min_id}"
+            )
+            messages = await client.get_messages(entity, min_id=min_id, limit=limit)
             for message in reversed(messages):
-                await self._handle_telegram_backfill_message(source, message)
                 count += 1
-        return count
+                if message.id <= min_id:
+                    self.log.trace(
+                        f"Skipping {message.id} in backfill response as it's lower than "
+                        f"the last bridged message ({min_id})"
+                    )
+                    continue
+                was_handled = await self._handle_telegram_backfill_message(source, message)
+                handled_count += 1 if was_handled else 0
+        return count, handled_count
 
     async def _handle_telegram_backfill_message(
         self, source: au.AbstractUser, msg: Message | MessageService
-    ) -> None:
+    ) -> bool:
         if msg.from_id and isinstance(msg.from_id, (PeerUser, PeerChannel)):
             sender = await p.Puppet.get_by_peer(msg.from_id)
         elif isinstance(msg.peer_id, PeerUser):
@@ -2885,14 +2930,17 @@ class Portal(DBPortal, BasePortal):
         if isinstance(msg, MessageService):
             if isinstance(msg.action, MessageActionContactSignUp):
                 await self.handle_telegram_joined(source, sender, msg, backfill=True)
+                return True
             else:
                 self.log.debug(
                     f"Unhandled service message {type(msg.action).__name__} in backfill"
                 )
         elif isinstance(msg, Message):
             await self.handle_telegram_message(source, sender, msg)
+            return True
         else:
             self.log.debug(f"Unhandled message type {type(msg).__name__} in backfill")
+        return False
 
     def _split_dm_reaction_counts(self, counts: list[ReactionCount]) -> list[MessagePeerReaction]:
         if len(counts) == 1:
@@ -2931,9 +2979,10 @@ class Portal(DBPortal, BasePortal):
         msg_id: TelegramID,
         data: MessageReactions,
         dbm: DBMessage | None = None,
+        timestamp: datetime | None = None,
     ) -> None:
         try:
-            await self.handle_telegram_reactions(source, msg_id, data, dbm)
+            await self.handle_telegram_reactions(source, msg_id, data, dbm, timestamp)
         except Exception:
             self.log.exception(f"Error handling reactions in message {msg_id}")
 
@@ -2943,6 +2992,7 @@ class Portal(DBPortal, BasePortal):
         msg_id: TelegramID,
         data: MessageReactions,
         dbm: DBMessage | None = None,
+        timestamp: datetime | None = None,
     ) -> None:
         if self.peer_type == "channel" and not self.megagroup:
             # We don't know who reacted in a channel, so we can't bridge it properly either
@@ -2971,10 +3021,16 @@ class Portal(DBPortal, BasePortal):
                 # recent_reactions = resp.reactions
 
         async with self.reaction_lock(dbm.mxid):
-            await self._handle_telegram_reactions_locked(dbm, recent_reactions, total_count)
+            await self._handle_telegram_reactions_locked(
+                dbm, recent_reactions, total_count, timestamp=timestamp
+            )
 
     async def _handle_telegram_reactions_locked(
-        self, msg: DBMessage, reaction_list: list[MessagePeerReaction], total_count: int
+        self,
+        msg: DBMessage,
+        reaction_list: list[MessagePeerReaction],
+        total_count: int,
+        timestamp: datetime | None = None,
     ) -> None:
         reactions = {
             p.Puppet.get_id_from_peer(reaction.peer_id): reaction.reaction
@@ -3002,7 +3058,7 @@ class Portal(DBPortal, BasePortal):
             self.log.debug(f"Bridging reaction {new_emoji} by {sender} to {msg.tgid}")
             puppet: p.Puppet = await p.Puppet.get_by_tgid(sender)
             mxid = await puppet.intent_for(self).react(
-                msg.mx_room, msg.mxid, variation_selector.add(new_emoji)
+                msg.mx_room, msg.mxid, variation_selector.add(new_emoji), timestamp=timestamp
             )
             await DBReaction(
                 mxid=mxid,
@@ -3028,7 +3084,7 @@ class Portal(DBPortal, BasePortal):
             intent = puppet.intent_for(self)
             await intent.redact(changed_reaction.mx_room, changed_reaction.mxid)
             changed_reaction.mxid = await intent.react(
-                msg.mx_room, msg.mxid, variation_selector.add(new_emoji)
+                msg.mx_room, msg.mxid, variation_selector.add(new_emoji), timestamp=timestamp
             )
             changed_reaction.reaction = new_emoji
             await changed_reaction.save()
@@ -3180,17 +3236,14 @@ class Portal(DBPortal, BasePortal):
             await dbm.insert()
             await DBMessage.replace_temp_mxid(temporary_identifier, self.mxid, event_id)
         except (IntegrityError, UniqueViolationError) as e:
-            self.log.exception(
-                f"{e.__class__.__name__} while saving message mapping. "
-                "This might mean that an update was handled after it left the "
-                "dedup cache queue. You can try enabling bridge.deduplication."
-                "pre_db_check in the config."
-            )
+            self.log.exception(f"{type(e).__name__} while saving message mapping")
             await intent.redact(self.mxid, event_id)
             return
         if isinstance(evt, Message) and evt.reactions:
             asyncio.create_task(
-                self.try_handle_telegram_reactions(source, dbm.tgid, evt.reactions, dbm=dbm)
+                self.try_handle_telegram_reactions(
+                    source, dbm.tgid, evt.reactions, dbm=dbm, timestamp=evt.date
+                )
             )
         await self._send_delivery_receipt(event_id)
 
@@ -3200,7 +3253,11 @@ class Portal(DBPortal, BasePortal):
         if source.is_relaybot and self.config["bridge.ignore_unbridged_group_chat"]:
             return False
         create_and_exit = (MessageActionChatCreate, MessageActionChannelCreate)
-        create_and_continue = (MessageActionChatAddUser, MessageActionChatJoinedByLink)
+        create_and_continue = (
+            MessageActionChatAddUser,
+            MessageActionChatJoinedByLink,
+            MessageActionChatJoinedByRequest,
+        )
         if isinstance(action, create_and_exit) or isinstance(action, create_and_continue):
             await self.create_matrix_room(
                 source, invites=[source.mxid], update_if_exists=isinstance(action, create_and_exit)
@@ -3230,7 +3287,7 @@ class Portal(DBPortal, BasePortal):
         elif isinstance(action, MessageActionChatAddUser):
             for user_id in action.users:
                 await self._add_telegram_user(TelegramID(user_id), source)
-        elif isinstance(action, MessageActionChatJoinedByLink):
+        elif isinstance(action, (MessageActionChatJoinedByLink, MessageActionChatJoinedByRequest)):
             await self._add_telegram_user(sender.id, source)
         elif isinstance(action, MessageActionChatDeleteUser):
             await self._delete_telegram_user(TelegramID(action.user_id), sender)
@@ -3257,6 +3314,15 @@ class Portal(DBPortal, BasePortal):
         backfill: bool = False,
     ) -> None:
         assert isinstance(update.action, MessageActionContactSignUp)
+
+        msg = await DBMessage.get_one_by_tgid(TelegramID(update.id), source.tgid)
+        if msg:
+            self.log.debug(
+                f"Ignoring new user message {update.id} (src {source.tgid}) as it was already "
+                f"handled into {msg.mxid}."
+            )
+            return
+
         content = TextMessageEventContent(msgtype=MessageType.EMOTE, body="joined Telegram")
         event_id = await self._send_message(
             sender.intent_for(self), content, timestamp=update.date
@@ -3268,10 +3334,7 @@ class Portal(DBPortal, BasePortal):
             tg_space=source.tgid,
             edit_index=0,
         ).insert()
-        # Automatically mark the notice as read if we're backfilling messages, mostly so that
-        # empty rooms created before the notice was added wouldn't become unread when the notice
-        # is backfilled in.
-        if backfill:
+        if self.config["bridge.always_read_joined_telegram_notice"]:
             double_puppet = await p.Puppet.get_by_tgid(source.tgid)
             if double_puppet and double_puppet.is_real_user:
                 await double_puppet.intent.mark_read(self.mxid, event_id)
@@ -3389,7 +3452,12 @@ class Portal(DBPortal, BasePortal):
             raise
 
     async def get_invite_link(
-        self, user: u.User, uses: int | None = None, expire: datetime | None = None
+        self,
+        user: u.User,
+        uses: int | None = None,
+        expire: datetime | None = None,
+        request_needed: bool = False,
+        title: str | None = None,
     ) -> str:
         if self.peer_type == "user":
             raise ValueError("You can't invite users to private chats.")
@@ -3397,7 +3465,11 @@ class Portal(DBPortal, BasePortal):
             return f"https://t.me/{self.username}"
         link = await user.client(
             ExportChatInviteRequest(
-                peer=await self.get_input_entity(user), expire_date=expire, usage_limit=uses
+                peer=await self.get_input_entity(user),
+                expire_date=expire,
+                usage_limit=uses,
+                request_needed=request_needed,
+                title=title,
             )
         )
         return link.link
@@ -3442,6 +3514,12 @@ class Portal(DBPortal, BasePortal):
             del self.by_mxid[self.mxid]
         except KeyError:
             pass
+        self.name_set = False
+        self.avatar_set = False
+        self.about = None
+        self.sponsored_event_id = None
+        self.sponsored_event_ts = None
+        self.sponsored_msg_random_id = None
         await super().delete()
         await DBMessage.delete_all(self.mxid)
         await DBReaction.delete_all(self.mxid)
@@ -3450,8 +3528,13 @@ class Portal(DBPortal, BasePortal):
     # endregion
     # region Class instance lookup
 
+    async def get_dm_puppet(self) -> p.Puppet | None:
+        if not self.is_direct:
+            return None
+        return await p.Puppet.get_by_tgid(self.tgid)
+
     async def postinit(self) -> None:
-        puppet = await p.Puppet.get_by_tgid(self.tgid) if self.is_direct else None
+        puppet = await self.get_dm_puppet()
         self._main_intent = puppet.intent_for(self) if self.is_direct else self.az.intent
 
         if self.tgid:

mautrix_telegram/portal_util/deduplication.py

@@ -62,7 +62,6 @@ media_content_table = {
 
 
 class PortalDedup:
-    pre_db_check: bool = False
     cache_queue_length: int = 256
 
     _dedup: deque[bytes | int]

mautrix_telegram/puppet.py

@@ -24,6 +24,7 @@ from telethon.tl.types import (
     ChatPhoto,
     ChatPhotoEmpty,
     InputPeerPhotoFileLocation,
+    InputPeerUser,
     PeerChannel,
     PeerChat,
     PeerUser,
@@ -72,6 +73,7 @@ class Puppet(DBPuppet, BasePuppet):
         displayname_quality: int = 0,
         disable_updates: bool = False,
         username: str | None = None,
+        phone: str | None = None,
         photo_id: str | None = None,
         avatar_url: ContentURI | None = None,
         name_set: bool = False,
@@ -92,6 +94,7 @@ class Puppet(DBPuppet, BasePuppet):
             displayname_quality=displayname_quality,
             disable_updates=disable_updates,
             username=username,
+            phone=phone,
             photo_id=photo_id,
             avatar_url=avatar_url,
             name_set=name_set,
@@ -128,6 +131,16 @@ class Puppet(DBPuppet, BasePuppet):
             PeerChannel(channel_id=self.tgid) if self.is_channel else PeerUser(user_id=self.tgid)
         )
 
+    @property
+    def contact_info(self) -> dict:
+        return {
+            "name": self.displayname,
+            "username": self.username,
+            "phone": f"+{self.phone.lstrip('+')}" if self.phone else None,
+            "is_bot": self.is_bot,
+            "avatar_url": self.avatar_url,
+        }
+
     @property
     def plain_displayname(self) -> str:
         return self.displayname_template.parse(self.displayname) or self.displayname
@@ -252,6 +265,10 @@ class Puppet(DBPuppet, BasePuppet):
             self.username = info.username
             changed = True
 
+        if getattr(info, "phone", None) and self.phone != info.phone:
+            self.phone = info.phone
+            changed = True
+
         if not self.disable_updates:
             try:
                 changed = await self.update_displayname(source, info) or changed
@@ -359,6 +376,7 @@ class Puppet(DBPuppet, BasePuppet):
                     location=InputPeerPhotoFileLocation(
                         peer=await self.get_input_entity(source), photo_id=photo.photo_id, big=True
                     ),
+                    async_upload=self.config["homeserver.async_media"],
                 )
                 if not file:
                     return False
@@ -413,7 +431,7 @@ class Puppet(DBPuppet, BasePuppet):
 
     @staticmethod
     def get_id_from_peer(peer: TypePeer | User | Channel) -> TelegramID:
-        if isinstance(peer, PeerUser):
+        if isinstance(peer, (PeerUser, InputPeerUser)):
             return TelegramID(peer.user_id)
         elif isinstance(peer, PeerChannel):
             return TelegramID(peer.channel_id)

mautrix_telegram/user.py

@@ -269,6 +269,13 @@ class User(DBUser, AbstractUser, BaseUser):
             return None
         return await pu.Puppet.get_by_tgid(self.tgid)
 
+    async def get_portal_with(self, puppet: pu.Puppet, create: bool = True) -> po.Portal | None:
+        if not self.tgid:
+            return None
+        return await po.Portal.get_by_tgid(
+            puppet.tgid, tg_receiver=self.tgid, peer_type="user" if create else None
+        )
+
     async def stop(self) -> None:
         if self._track_connection_task:
             self._track_connection_task.cancel()
@@ -372,7 +379,7 @@ class User(DBUser, AbstractUser, BaseUser):
         if not self.config["bridge.kick_on_logout"]:
             return
         portals = await self.get_cached_portals()
-        for _, portal in portals.values():
+        for portal in portals.values():
             if not portal or portal.deleted or not portal.mxid or portal.has_bot:
                 continue
             if portal.peer_type == "user":
@@ -462,17 +469,22 @@ 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}")
             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}")
             await puppet.intent.remove_room_tag(portal.mxid, tag)
 
-    async def _mute_room(cls, puppet: pu.Puppet, portal: po.Portal, mute_until: datetime) -> None:
-        if not cls.config["bridge.mute_bridging"] or not portal or not portal.mxid:
+    async def _mute_room(self, puppet: pu.Puppet, portal: po.Portal, mute_until: datetime) -> None:
+        if not self.config["bridge.mute_bridging"] or not portal or not portal.mxid:
             return
         now = datetime.utcnow().replace(tzinfo=timezone.utc)
         if mute_until is not None and mute_until > now:
+            self.log.debug(
+                f"Muting {portal.mxid}/{portal.tgid} (muted until {mute_until} on Telegram)"
+            )
             await puppet.intent.set_push_rule(
                 PushRuleScope.GLOBAL,
                 PushRuleKind.ROOM,
@@ -484,6 +496,7 @@ class User(DBUser, AbstractUser, BaseUser):
                 await puppet.intent.remove_push_rule(
                     PushRuleScope.GLOBAL, PushRuleKind.ROOM, portal.mxid
                 )
+                self.log.debug(f"Unmuted {portal.mxid}/{portal.tgid}")
             except MNotFound:
                 pass
 
@@ -646,20 +659,28 @@ class User(DBUser, AbstractUser, BaseUser):
             acc = (acc * 20261 + contact) & 0xFFFFFFFF
         return acc & 0x7FFFFFFF
 
-    async def sync_contacts(self) -> None:
+    async def sync_contacts(self, get_info: bool = False) -> dict[TelegramID, dict]:
         existing_contacts = await self.get_contacts()
         contact_hash = self._hash_contacts(self.saved_contacts, existing_contacts)
         response = await self.client(GetContactsRequest(hash=contact_hash))
         if isinstance(response, ContactsNotModified):
-            return
+            if get_info:
+                return {
+                    tgid: (await pu.Puppet.get_by_tgid(tgid)).contact_info
+                    for tgid in existing_contacts
+                }
+            return {}
         self.log.debug(f"Updating contacts of {self.name}...")
         if self.saved_contacts != response.saved_count:
             self.saved_contacts = response.saved_count
             await self.save()
+        contacts = {}
         for user in response.users:
-            puppet = await pu.Puppet.get_by_tgid(user.id)
+            puppet: pu.Puppet = await pu.Puppet.get_by_tgid(user.id)
             await puppet.update_info(self, user)
-        await self.set_contacts(user.id for user in response.users)
+            contacts[user.id] = puppet.contact_info
+        await self.set_contacts(contacts.keys())
+        return contacts
 
     # endregion
     # region Class instance lookup

mautrix_telegram/util/file_transfer.py

@@ -144,6 +144,7 @@ async def transfer_thumbnail_to_matrix(
     custom_data: bytes | None = None,
     width: int | None = None,
     height: int | None = None,
+    async_upload: bool = False,
 ) -> DBTelegramFile | None:
     if not Image or not VideoFileClip:
         return None
@@ -178,7 +179,7 @@ async def transfer_thumbnail_to_matrix(
     if encrypt:
         file, decryption_info = encrypt_attachment(file)
         upload_mime_type = "application/octet-stream"
-    content_uri = await intent.upload_media(file, upload_mime_type)
+    content_uri = await intent.upload_media(file, upload_mime_type, async_upload=async_upload)
     if decryption_info:
         decryption_info.url = content_uri
 
@@ -220,6 +221,7 @@ async def transfer_file_to_matrix(
     filename: str | None = None,
     encrypt: bool = False,
     parallel_id: int | None = None,
+    async_upload: bool = False,
 ) -> DBTelegramFile | None:
     location_id = _location_to_id(location)
     if not location_id:
@@ -246,6 +248,7 @@ async def transfer_file_to_matrix(
             filename,
             encrypt,
             parallel_id,
+            async_upload=async_upload,
         )
 
 
@@ -260,6 +263,7 @@ async def _unlocked_transfer_file_to_matrix(
     filename: str | None,
     encrypt: bool,
     parallel_id: int | None,
+    async_upload: bool = False,
 ) -> DBTelegramFile | None:
     db_file = await DBTelegramFile.get(loc_id)
     if db_file:
@@ -305,7 +309,7 @@ async def _unlocked_transfer_file_to_matrix(
         if encrypt and encrypt_attachment:
             file, decryption_info = encrypt_attachment(file)
             upload_mime_type = "application/octet-stream"
-        content_uri = await intent.upload_media(file, upload_mime_type)
+        content_uri = await intent.upload_media(file, upload_mime_type, async_upload=async_upload)
         if decryption_info:
             decryption_info.url = content_uri
 
@@ -325,7 +329,13 @@ async def _unlocked_transfer_file_to_matrix(
             thumbnail = thumbnail.location
         try:
             db_file.thumbnail = await transfer_thumbnail_to_matrix(
-                client, intent, thumbnail, video=file, mime_type=mime_type, encrypt=encrypt
+                client,
+                intent,
+                thumbnail,
+                video=file,
+                mime_type=mime_type,
+                encrypt=encrypt,
+                async_upload=async_upload,
             )
         except FileIdInvalidError:
             log.warning(f"Failed to transfer thumbnail for {thumbnail!s}", exc_info=True)
@@ -340,6 +350,7 @@ async def _unlocked_transfer_file_to_matrix(
             mime_type=converted_anim.thumbnail_mime,
             width=converted_anim.width,
             height=converted_anim.height,
+            async_upload=async_upload,
         )
 
     try:

mautrix_telegram/web/provisioning/__init__.py

@@ -21,7 +21,7 @@ import json
 import logging
 
 from aiohttp import web
-from telethon.tl.types import ChannelForbidden, ChatForbidden, TypeChat
+from telethon.tl.types import ChannelForbidden, ChatForbidden, TypeChat, User as TLUser
 from telethon.utils import get_peer_id, resolve_id
 
 from mautrix.appservice import AppService
@@ -53,18 +53,20 @@ class ProvisioningAPI(AuthAPI):
 
         self.app = web.Application(loop=bridge.loop, middlewares=[self.error_middleware])
 
-        portal_prefix = "/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", "/portal/{tgid:-[0-9]+}", self.get_portal_by_tgid)
+        self.app.router.add_route("GET", "/v1/portal/{tgid:-[0-9]+}", self.get_portal_by_tgid)
         self.app.router.add_route(
             "POST", portal_prefix + "/connect/{chat_id:-[0-9]+}", self.connect_chat
         )
         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 = "/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)
+        self.app.router.add_route("POST", f"{user_prefix}/pm/{{identifier}}", self.start_dm)
 
         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)
@@ -212,7 +214,7 @@ class ProvisioningAPI(AuthAPI):
             portal.photo_id = ""
             await portal.save()
 
-        asyncio.create_task(portal.update_matrix_room(user, entity, direct=False, levels=levels))
+        asyncio.create_task(portal.update_matrix_room(user, entity, levels=levels))
 
         return web.Response(status=202, body="{}")
 
@@ -393,6 +395,62 @@ class ProvisioningAPI(AuthAPI):
                 ]
             )
 
+    async def get_contacts(self, request: web.Request) -> web.Response:
+        data, user, err = await self.get_user_request_info(request, expect_logged_in=True)
+        if err is not None:
+            return err
+        return web.json_response(data=await user.sync_contacts())
+
+    async def start_dm(self, request: web.Request) -> web.Response:
+        data, user, err = await self.get_user_request_info(request, expect_logged_in=True)
+        if err is not None:
+            return err
+        try:
+            identifier: str | int = request.match_info["identifier"]
+            if isinstance(identifier, str) and identifier.isdecimal():
+                identifier = int(identifier)
+            target = await user.client.get_entity(identifier)
+        except ValueError:
+            return web.json_response(
+                {
+                    "error": "Invalid user identifier or user not found.",
+                    "errcode": "M_NOT_FOUND",
+                },
+                status=404,
+            )
+
+        if not target:
+            return web.json_response(
+                {
+                    "error": "User not found.",
+                    "errcode": "M_NOT_FOUND",
+                },
+                status=404,
+            )
+        elif not isinstance(target, TLUser):
+            return web.json_response(
+                {
+                    "error": "Identifier is not a user.",
+                },
+                status=400,
+            )
+        portal = await Portal.get_by_entity(target, tg_receiver=user.tgid)
+        puppet = await portal.get_dm_puppet()
+        if portal.mxid:
+            just_created = False
+        else:
+            await portal.create_matrix_room(user, target, [user.mxid])
+            just_created = True
+        return web.json_response(
+            {
+                "room_id": portal.mxid,
+                "just_created": just_created,
+                "id": portal.tgid,
+                "contact_info": puppet.contact_info,
+            },
+            status=201 if just_created else 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:
@@ -574,7 +632,7 @@ class ProvisioningAPI(AuthAPI):
         data = None
         if want_data and (request.method == "POST" or request.method == "PUT"):
             data = await self.get_data(request)
-            if not data:
+            if data is None:
                 return (
                     None,
                     None,

optional-requirements.txt

@@ -18,7 +18,7 @@ moviepy>=1,<2
 phonenumbers>=8,<9
 
 #/metrics
-prometheus_client>=0.6,<0.14
+prometheus_client>=0.6,<0.15
 
 #/e2be
 python-olm>=3,<4

pyproject.toml

@@ -4,9 +4,9 @@ force_to_top = "typing"
 from_first = true
 combine_as_imports = true
 known_first_party = "mautrix"
+known_third_party = "telethon"
 line_length = 99
 
 [tool.black]
 line-length = 99
 target-version = ["py38"]
-required-version = "22.1.0"

requirements.txt

@@ -3,10 +3,10 @@ python-magic>=0.4,<0.5
 commonmark>=0.8,<0.10
 aiohttp>=3,<4
 yarl>=1,<2
-mautrix>=0.14.9,<0.15
+mautrix>=0.16.0,<0.17
 #telethon>=1.24,<1.25
 # Fork to make session storage async and update to layer 138
-tulir-telethon==1.25.0a5
+tulir-telethon==1.25.0a7
 asyncpg>=0.20,<0.26
 mako>=1,<2
 setuptools

setup.py

@@ -51,7 +51,7 @@ setuptools.setup(
 
     install_requires=install_requires,
     extras_require=extras_require,
-    python_requires="~=3.7",
+    python_requires="~=3.8",
 
     classifiers=[
         "Development Status :: 4 - Beta",