diff --git a/ROADMAP.md b/ROADMAP.md index 336fb935..ee4e5608 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -3,7 +3,7 @@ * Matrix → Telegram * [x] Message content (text, formatting, files, etc..) * [x] Message redactions - * [ ] Message reactions + * [x] Message reactions * [x] Message edits * [ ] ‡ Message history * [x] Presence diff --git a/mautrix_telegram/db/message.py b/mautrix_telegram/db/message.py index f1625fbc..cee5abb2 100644 --- a/mautrix_telegram/db/message.py +++ b/mautrix_telegram/db/message.py @@ -143,6 +143,11 @@ class Message: q = "UPDATE message SET mxid=$1 WHERE mxid=$2 AND mx_room=$3" await cls.db.execute(q, real_mxid, temp_mxid, mx_room) + @classmethod + async def delete_temp_mxid(cls, temp_mxid: str, mx_room: RoomID) -> None: + q = "DELETE FROM message WHERE mxid=$1 AND mx_room=$2" + await cls.db.execute(q, temp_mxid, mx_room) + @property def _values(self): return ( diff --git a/mautrix_telegram/db/reaction.py b/mautrix_telegram/db/reaction.py index 71c49dd7..9880489f 100644 --- a/mautrix_telegram/db/reaction.py +++ b/mautrix_telegram/db/reaction.py @@ -55,6 +55,13 @@ class Reaction: q = f"SELECT {cls.columns} FROM reaction WHERE mxid=$1 AND mx_room=$2" return cls._from_row(await cls.db.fetchrow(q, mxid, mx_room)) + @classmethod + async def get_by_sender( + cls, mxid: EventID, mx_room: RoomID, tg_sender: TelegramID + ) -> Reaction | None: + q = f"SELECT {cls.columns} FROM reaction WHERE msg_mxid=$1 AND mx_room=$2 AND tg_sender=$3" + return cls._from_row(await cls.db.fetchrow(q, mxid, mx_room, tg_sender)) + @classmethod async def get_all_by_message(cls, mxid: EventID, mx_room: RoomID) -> list[Reaction]: q = f"SELECT {cls.columns} FROM reaction WHERE msg_mxid=$1 AND mx_room=$2" diff --git a/mautrix_telegram/matrix.py b/mautrix_telegram/matrix.py index 7ff88c59..252926e1 100644 --- a/mautrix_telegram/matrix.py +++ b/mautrix_telegram/matrix.py @@ -15,9 +15,8 @@ # along with this program. If not, see . from __future__ import annotations -from typing import TYPE_CHECKING, Iterable +from typing import TYPE_CHECKING -from mautrix.appservice import DOUBLE_PUPPET_SOURCE_KEY from mautrix.bridge import BaseMatrixHandler from mautrix.errors import MatrixError from mautrix.types import ( @@ -28,9 +27,8 @@ from mautrix.types import ( MessageType, PresenceEvent, PresenceState, + ReactionEvent, ReceiptEvent, - ReceiptEventContent, - ReceiptType, RedactionEvent, RoomAvatarStateEventContent as AvatarContent, RoomID, @@ -278,6 +276,20 @@ class MatrixHandler(BaseMatrixHandler): await portal.handle_matrix_deletion(sender, evt.redacts, evt.event_id) + @staticmethod + async def handle_reaction(evt: ReactionEvent) -> None: + sender = await u.User.get_and_start_by_mxid(evt.sender) + if not await sender.has_full_access(): + return + + portal = await po.Portal.get_by_mxid(evt.room_id) + if not portal or not portal.allow_bridging: + return + + await portal.handle_matrix_reaction( + sender, evt.content.relates_to.event_id, evt.content.relates_to.key, evt.event_id + ) + @staticmethod async def handle_power_levels(evt: StateEvent) -> None: portal = await po.Portal.get_by_mxid(evt.room_id) @@ -400,6 +412,8 @@ class MatrixHandler(BaseMatrixHandler): async def handle_event(self, evt: Event) -> None: if evt.type == EventType.ROOM_REDACTION: await self.handle_redaction(evt) + elif evt.type == EventType.REACTION: + await self.handle_reaction(evt) async def handle_state_event(self, evt: StateEvent) -> None: if evt.type == EventType.ROOM_POWER_LEVELS: diff --git a/mautrix_telegram/portal.py b/mautrix_telegram/portal.py index 1b55798c..1b82332c 100644 --- a/mautrix_telegram/portal.py +++ b/mautrix_telegram/portal.py @@ -45,6 +45,7 @@ from telethon.errors import ( PhotoExtInvalidError, PhotoInvalidDimensionsError, PhotoSaveFileInvalidError, + ReactionInvalidError, RPCError, ) from telethon.tl.functions.channels import ( @@ -65,6 +66,7 @@ from telethon.tl.functions.messages import ( ExportChatInviteRequest, GetMessageReactionsListRequest, MigrateChatRequest, + SendReactionRequest, SetTypingRequest, UnpinAllMessagesRequest, UpdatePinnedMessageRequest, @@ -216,6 +218,10 @@ TypeChatPhoto = Union[ChatPhoto, ChatPhotoEmpty, Photo, PhotoEmpty] MediaHandler = Callable[["au.AbstractUser", IntentAPI, Message, RelatesTo], Awaitable[EventID]] +class BridgingError(Exception): + pass + + class DocAttrs(NamedTuple): name: str | None mime_type: str | None @@ -1516,7 +1522,7 @@ class Portal(DBPortal, BasePortal): else: if content.file: if not decrypt_attachment: - raise Exception( + raise BridgingError( f"Can't bridge encrypted media event {event_id}: " "encryption dependencies not installed" ) @@ -1746,7 +1752,7 @@ class Portal(DBPortal, BasePortal): bridge_notices = self.get_config("bridge_notices.default") excepted = sender.mxid in self.get_config("bridge_notices.exceptions") if not bridge_notices and not excepted: - raise Exception("Notices are not configured to be bridged.") + raise BridgingError("Notices are not configured to be bridged.") if content.msgtype in (MessageType.TEXT, MessageType.EMOTE, MessageType.NOTICE): await self._pre_process_matrix_message(sender, not logged_in, content) @@ -1779,7 +1785,7 @@ class Portal(DBPortal, BasePortal): f"Didn't handle Matrix event {event_id} due to unknown msgtype {content.msgtype}" ) self.log.trace("Unhandled Matrix event content: %s", content) - raise Exception(f"Unhandled msgtype {content.msgtype}") + raise BridgingError(f"Unhandled msgtype {content.msgtype}") async def handle_matrix_unpin_all(self, sender: u.User, pin_event_id: EventID) -> None: await sender.client(UnpinAllMessagesRequest(peer=self.peer)) @@ -1809,9 +1815,12 @@ class Portal(DBPortal, BasePortal): ) -> None: try: await self._handle_matrix_deletion(deleter, event_id) - except Exception as e: + except BridgingError as e: self.log.debug(str(e)) await self._send_bridge_error(deleter, e, redaction_event_id, EventType.ROOM_REDACTION) + except Exception as e: + self.log.exception(f"Failed to bridge redaction by {deleter.mxid}") + await self._send_bridge_error(deleter, e, redaction_event_id, EventType.ROOM_REDACTION) else: deleter.send_remote_checkpoint( MessageSendCheckpointStatus.SUCCESS, @@ -1821,26 +1830,102 @@ class Portal(DBPortal, BasePortal): ) await self._send_delivery_receipt(redaction_event_id) + async def _handle_matrix_reaction_deletion( + self, deleter: u.User, event_id: EventID, tg_space: TelegramID + ) -> None: + reaction = await DBReaction.get_by_mxid(event_id, self.mxid) + if not reaction: + raise BridgingError(f"Ignoring Matrix redaction of unknown event {event_id}") + elif reaction.tg_sender != deleter.tgid: + raise BridgingError(f"Ignoring Matrix redaction of reaction by another user") + reaction_target = await DBMessage.get_by_mxid( + reaction.msg_mxid, reaction.mx_room, tg_space + ) + if not reaction_target or reaction_target.redacted: + raise BridgingError( + f"Ignoring Matrix redaction of reaction to unknown event {reaction.msg_mxid}" + ) + async with self.reaction_lock(reaction_target.mxid): + await reaction.delete() + await deleter.client(SendReactionRequest(peer=self.peer, msg_id=reaction_target.tgid)) + async def _handle_matrix_deletion(self, deleter: u.User, event_id: EventID) -> None: real_deleter = deleter if not await deleter.needs_relaybot(self) else self.bot - space = self.tgid if self.peer_type == "channel" else real_deleter.tgid - message = await DBMessage.get_by_mxid(event_id, self.mxid, space) + tg_space = self.tgid if self.peer_type == "channel" else real_deleter.tgid + message = await DBMessage.get_by_mxid(event_id, self.mxid, tg_space) if not message: - raise Exception(f"Ignoring Matrix redaction of unknown event {event_id}") + await self._handle_matrix_reaction_deletion(real_deleter, event_id, tg_space) elif message.redacted: - raise Exception( + raise BridgingError( "Ignoring Matrix redaction of already redacted event " f"{message.mxid} in {message.mx_room}" ) elif message.edit_index != 0: await message.mark_redacted() - raise Exception( + raise BridgingError( f"Ignoring Matrix redaction of edit event {message.mxid} in {message.mx_room}" ) else: await message.mark_redacted() await real_deleter.client.delete_messages(self.peer, [message.tgid]) + async def handle_matrix_reaction( + self, user: u.User, target_event_id: EventID, reaction: str, reaction_event_id: EventID + ) -> None: + try: + async with self.reaction_lock(target_event_id): + await self._handle_matrix_reaction( + user, target_event_id, reaction, reaction_event_id + ) + except BridgingError as e: + self.log.debug(str(e)) + await self._send_bridge_error(user, e, reaction_event_id, EventType.REACTION) + except ReactionInvalidError as e: + await self.main_intent.redact(self.mxid, reaction_event_id, reason="Emoji not allowed") + self.log.debug(f"Failed to bridge reaction by {user.mxid}: emoji not allowed") + await self._send_bridge_error(user, e, reaction_event_id, EventType.REACTION) + except Exception as e: + self.log.exception(f"Failed to bridge reaction by {user.mxid}") + await self._send_bridge_error(user, e, reaction_event_id, EventType.REACTION) + else: + user.send_remote_checkpoint( + MessageSendCheckpointStatus.SUCCESS, + reaction_event_id, + self.mxid, + EventType.REACTION, + ) + await self._send_delivery_receipt(reaction_event_id) + + async def _handle_matrix_reaction( + self, user: u.User, target_event_id: EventID, emoji: str, reaction_event_id: EventID + ) -> None: + tg_space = self.tgid if self.peer_type == "channel" else user.tgid + msg = await DBMessage.get_by_mxid(target_event_id, self.mxid, tg_space) + if not msg: + raise BridgingError(f"Ignoring Matrix reaction to unknown event {target_event_id}") + elif msg.redacted: + raise BridgingError(f"Ignoring Matrix reaction to redacted event {target_event_id}") + elif msg.edit_index != 0: + raise BridgingError(f"Ignoring Matrix reaction to edit event {target_event_id}") + + emoji = variation_selector.remove(emoji) + existing_react = await DBReaction.get_by_sender(msg.mxid, msg.mx_room, user.tgid) + await user.client(SendReactionRequest(peer=self.peer, msg_id=msg.tgid, reaction=emoji)) + if existing_react: + puppet = await user.get_puppet() + await puppet.intent_for(self).redact(existing_react.mx_room, existing_react.mxid) + existing_react.mxid = reaction_event_id + existing_react.reaction = emoji + await existing_react.save() + else: + await DBReaction( + mxid=reaction_event_id, + mx_room=self.mxid, + msg_mxid=msg.mxid, + tg_sender=user.tgid, + reaction=emoji, + ).save() + async def _update_telegram_power_level( self, sender: u.User, user_id: TelegramID, level: int ) -> None: @@ -2424,7 +2509,11 @@ class Portal(DBPortal, BasePortal): prev_edit_msg = await DBMessage.get_one_by_tgid( TelegramID(evt.id), tg_space, edit_index=-1 ) - if not prev_edit_msg: + if ( + not prev_edit_msg + or prev_edit_msg.mxid == mxid + or prev_edit_msg.content_hash == event_hash + ): return await DBMessage( mxid=mxid, @@ -2454,6 +2543,7 @@ class Portal(DBPortal, BasePortal): f"Ignoring edit of message {evt.id}@{tg_space} (src {source.tgid}):" " content hash didn't change" ) + await DBMessage.delete_temp_mxid(temporary_identifier, self.mxid) return content.msgtype = ( diff --git a/requirements.txt b/requirements.txt index 9ac5c726..d7a4cbe1 100644 --- a/requirements.txt +++ b/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.0,<0.15 +mautrix>=0.14.1,<0.15 #telethon>=1.24,<1.25 -# Fork to make session storage async and update to layer 136 -tulir-telethon==1.25.0a2 +# Fork to make session storage async and update to layer 137 +tulir-telethon==1.25.0a3 asyncpg>=0.20,<0.26 mako>=1,<2 setuptools