Bump version to 0.3.0rc3

* Add pre-send message database check for deduplication
* Make dedup cache queue length configurable

example-config.yaml

@@ -17,6 +17,9 @@ appservice:
     # The hostname and port where this appservice should listen.
     hostname: 0.0.0.0
     port: 8080
+    # The maximum body size of appservice API requests (from the homeserver) in mebibytes
+    # Usually 1 is enough, but on high-traffic bridges you might need to increase this to avoid 413s
+    max_body_size: 1
 
     # The full URI to the database. SQLite and Postgres are fully supported.
     # Other DBMSes supported by SQLAlchemy may or may not work.
@@ -133,6 +136,16 @@ bridge:
     # your own Matrix account as the Matrix puppet for your Telegram account.
     sync_with_custom_puppets: true
 
+    # Some config options related to Telegram message deduplication.
+    # The default values are usually fine, but some debug messages/warnings might recommend you
+    # change these.
+    deduplication:
+        # Whether or not to check the database if the message about to be sent is a duplicate.
+        pre_db_check: false
+        # The number of latest events to keep when checking for duplicates.
+        # You might need to increase this on high-traffic bridge instances.
+        cache_queue_length: 20
+
     # The formats to use when sending messages to Telegram via the relay bot.
     #
     # Telegram doesn't have built-in emotes, so the m.emote format is also used for non-relaybot users.

mautrix_telegram/__init__.py

@@ -1,2 +1,2 @@
-__version__ = "0.3.0rc1"
+__version__ = "0.3.0rc3"
 __author__ = "Tulir Asokan <tulir@maunium.net>"

mautrix_telegram/__main__.py

@@ -82,11 +82,15 @@ session_container = AlchemySessionContainer(engine=db_engine, session=db_session
 loop = asyncio.get_event_loop()  # type: asyncio.AbstractEventLoop
 
 state_store = SQLStateStore(db_session)
+mebibyte = 1024 ** 2
 appserv = AppService(config["homeserver.address"], config["homeserver.domain"],
                      config["appservice.as_token"], config["appservice.hs_token"],
                      config["appservice.bot_username"], log="mau.as", loop=loop,
                      verify_ssl=config["homeserver.verify_ssl"], state_store=state_store,
-                     real_user_content_key="net.maunium.telegram.puppet")
+                     real_user_content_key="net.maunium.telegram.puppet",
+                     aiohttp_params={
+                         "client_max_size": config["appservice.max_body_size"] * mebibyte
+                     })
 
 context = Context(appserv, db_session, config, loop, session_container)
 
@@ -101,7 +105,6 @@ if config["appservice.provisioning.enabled"]:
                            provisioning_api.app)
     context.provisioning_api = provisioning_api
 
-
 with appserv.run(config["appservice.hostname"], config["appservice.port"]) as start:
     init_db(db_session)
     init_abstract_user(context)

mautrix_telegram/config.py

@@ -153,6 +153,7 @@ class Config(DictWithRecursion):
             copy("appservice.address")
         copy("appservice.hostname")
         copy("appservice.port")
+        copy("appservice.max_body_size")
 
         copy("appservice.database")
 
@@ -195,6 +196,9 @@ class Config(DictWithRecursion):
         copy("bridge.catch_up")
         copy("bridge.sync_with_custom_puppets")
 
+        copy("bridge.deduplication.pre_db_check")
+        copy("bridge.deduplication.cache_queue_length")
+
         if "bridge.message_formats.m_text" in self:
             del self["bridge.message_formats"]
         copy_dict("bridge.message_formats", override_existing_map=False)

mautrix_telegram/matrix.py

@@ -386,7 +386,8 @@ class MatrixHandler:
         elif evt_type == "m.room.redaction":
             await self.handle_redaction(room_id, sender, evt["redacts"])
         elif evt_type == "m.room.power_levels":
-            await self.handle_power_levels(room_id, sender, evt["content"], evt["prev_content"])
+            prev_content = evt.get("unsigned", {}).get("prev_content", {})  # type: dict
+            await self.handle_power_levels(room_id, sender, evt["content"], prev_content)
         elif evt_type in ("m.room.name", "m.room.avatar", "m.room.topic"):
             await self.handle_room_meta(evt_type, room_id, sender, evt["content"])
         elif evt_type == "m.room.pinned_events":

mautrix_telegram/portal.py

@@ -66,12 +66,19 @@ class Portal:
     az = None  # type: AppService
     bot = None  # type: Bot
     loop = None  # type: asyncio.AbstractEventLoop
+
     filter_mode = None  # type: str
     filter_list = None  # type: List[str]
+
     bridge_notices = False  # type: bool
+
+    dedup_pre_db_check = False  # type: bool
+    dedup_cache_queue_length = 20  # type: int
+
     alias_template = None  # type: str
     mx_alias_regex = None  # type: Pattern
     hs_domain = None  # type: str
+
     by_mxid = {}  # type: Dict[str, Portal]
     by_tgid = {}  # type: Dict[Tuple[int, int], Portal]
 
@@ -191,7 +198,7 @@ class Portal:
 
         self._dedup_action.append(evt_hash)
 
-        if len(self._dedup_action) > 20:
+        if len(self._dedup_action) > self.dedup_cache_queue_length:
             self._dedup_action.popleft()
         return False
 
@@ -221,7 +228,7 @@ class Portal:
         self._dedup_mxid[evt_hash] = mxid
         self._dedup.append(evt_hash)
 
-        if len(self._dedup) > 20:
+        if len(self._dedup) > self.dedup_cache_queue_length:
             del self._dedup_mxid[self._dedup.popleft()]
         return None
 
@@ -1330,6 +1337,8 @@ class Portal:
 
         msg = DBMessage.query.get((evt.id, tg_space))
         if not msg:
+            self.log.info(f"Didn't find edited message {evt.id}@{tg_space} (src {source.tgid}) "
+                          "in database.")
             # Oh crap
             return
         msg.mxid = mxid
@@ -1363,6 +1372,15 @@ class Portal:
                 self.db.commit()
             return
 
+        if self.dedup_pre_db_check and self.peer_type == "channel":
+            msg = DBMessage.query.get((evt.id, tg_space))
+            if msg:
+                self.log.debug(f"Ignoring message {evt.id} (src {source.tgid}) as it was already"
+                               f"handled into {msg.mxid}. This duplicate was catched in the db "
+                               "check. If you get this message often, consider increasing"
+                               "bridge.deduplication.cache_queue_length in the config.")
+                return
+
         if sender and not sender.displayname:
             self.log.debug(f"Telegram user {sender.tgid} sent a message, but doesn't have a"
                            "displayname, updating info...")
@@ -1416,10 +1434,16 @@ class Portal:
                         DBMessage.mxid == temporary_identifier) \
                 .update({"mxid": mxid})
         except FlushError as e:
-            self.log.exception(f"{e.__class__.__name__} while saving message mapping.")
+            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.")
             await intent.redact(self.mxid, mxid)
         except (IntegrityError, InvalidRequestError) as e:
-            self.log.exception(f"{e.__class__.__name__} while saving message mapping.")
+            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.db.rollback()
             await intent.redact(self.mxid, mxid)
 
@@ -1740,6 +1764,8 @@ def init(context: Context):
     Portal.bridge_notices = config["bridge.bridge_notices"]
     Portal.filter_mode = config["bridge.filter.mode"]
     Portal.filter_list = config["bridge.filter.list"]
+    Portal.dedup_pre_db_check = config["bridge.deduplication.pre_db_check"]
+    Portal.dedup_cache_queue_length = config["bridge.deduplication.cache_queue_length"]
     Portal.alias_template = config.get("bridge.alias_template", "telegram_{groupname}")
     Portal.hs_domain = config["homeserver.domain"]
     Portal.mx_alias_regex = re.compile(

setup.py

@@ -21,19 +21,20 @@ setuptools.setup(
 
     description="A Matrix-Telegram hybrid puppeting/relaybot bridge.",
     long_description=open("README.md").read(),
+    long_description_content_type="text/markdown",
 
     packages=setuptools.find_packages(),
 
     install_requires=[
         "aiohttp>=3.0.1,<4",
-        "mautrix-appservice>=0.3.1,<0.4.0",
+        "mautrix-appservice>=0.3.6,<0.4.0",
         "SQLAlchemy>=1.2.3,<2",
         "alembic>=1.0.0,<2",
         "Markdown>=2.6.11,<3",
         "ruamel.yaml>=0.15.35,<0.16",
         "future-fstrings>=0.4.2",
         "python-magic>=0.4.15,<0.5",
-        "telethon>=1.0,<1.1",
+        "telethon>=1.0,<1.2",
         "telethon-session-sqlalchemy>=0.2.3,<0.3",
     ],
     extras_require=extras,