portal.py 71 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818
  1. # mautrix-instagram - A Matrix-Instagram puppeting bridge.
  2. # Copyright (C) 2022 Tulir Asokan
  3. #
  4. # This program is free software: you can redistribute it and/or modify
  5. # it under the terms of the GNU Affero General Public License as published by
  6. # the Free Software Foundation, either version 3 of the License, or
  7. # (at your option) any later version.
  8. #
  9. # This program is distributed in the hope that it will be useful,
  10. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. # GNU Affero General Public License for more details.
  13. #
  14. # You should have received a copy of the GNU Affero General Public License
  15. # along with this program. If not, see <https://www.gnu.org/licenses/>.
  16. from __future__ import annotations
  17. from typing import TYPE_CHECKING, Any, AsyncGenerator, Awaitable, Callable, Union, cast
  18. from collections import deque
  19. from io import BytesIO
  20. import asyncio
  21. import json
  22. import mimetypes
  23. import re
  24. import time
  25. import asyncpg
  26. import magic
  27. from mauigpapi.types import (
  28. AnimatedMediaItem,
  29. CommandResponse,
  30. ExpiredMediaItem,
  31. MediaShareItem,
  32. MediaType,
  33. MessageSyncMessage,
  34. Reaction,
  35. ReactionStatus,
  36. ReelMediaShareItem,
  37. ReelShareType,
  38. RegularMediaItem,
  39. Thread,
  40. ThreadItem,
  41. ThreadItemType,
  42. ThreadUser,
  43. ThreadUserLastSeenAt,
  44. TypingStatus,
  45. VoiceMediaItem,
  46. )
  47. from mautrix.appservice import AppService, IntentAPI
  48. from mautrix.bridge import BasePortal, NotificationDisabler, async_getter_lock
  49. from mautrix.errors import MatrixError, MForbidden, MNotFound, SessionNotFound
  50. from mautrix.types import (
  51. AudioInfo,
  52. BeeperMessageStatusEventContent,
  53. ContentURI,
  54. EventID,
  55. EventType,
  56. Format,
  57. ImageInfo,
  58. LocationMessageEventContent,
  59. MediaMessageEventContent,
  60. MessageEventContent,
  61. MessageStatusReason,
  62. MessageType,
  63. RelatesTo,
  64. RelationType,
  65. RoomID,
  66. TextMessageEventContent,
  67. UserID,
  68. VideoInfo,
  69. )
  70. from mautrix.util import ffmpeg
  71. from mautrix.util.message_send_checkpoint import MessageSendCheckpointStatus
  72. from mautrix.util.simple_lock import SimpleLock
  73. from . import matrix as m, puppet as p, user as u
  74. from .config import Config
  75. from .db import Message as DBMessage, Portal as DBPortal, Reaction as DBReaction
  76. if TYPE_CHECKING:
  77. from .__main__ import InstagramBridge
  78. try:
  79. from mautrix.crypto.attachments import decrypt_attachment, encrypt_attachment
  80. except ImportError:
  81. encrypt_attachment = decrypt_attachment = None
  82. try:
  83. from PIL import Image
  84. except ImportError:
  85. Image = None
  86. StateBridge = EventType.find("m.bridge", EventType.Class.STATE)
  87. StateHalfShotBridge = EventType.find("uk.half-shot.bridge", EventType.Class.STATE)
  88. MediaData = Union[
  89. AnimatedMediaItem,
  90. ExpiredMediaItem,
  91. MediaShareItem,
  92. ReelMediaShareItem,
  93. RegularMediaItem,
  94. VoiceMediaItem,
  95. ]
  96. MediaUploadFunc = Callable[["u.User", MediaData, IntentAPI], Awaitable[MediaMessageEventContent]]
  97. # This doesn't need to capture all valid URLs, it's enough to catch most of them.
  98. # False negatives simply mean the link won't be linkified on Instagram,
  99. # but false positives will cause the message to fail to send.
  100. SIMPLE_URL_REGEX = re.compile(
  101. r"(?P<url>https?://[\da-z.-]+\.[a-z]{2,}(?:/[^\s]*)?)", flags=re.IGNORECASE
  102. )
  103. class Portal(DBPortal, BasePortal):
  104. by_mxid: dict[RoomID, Portal] = {}
  105. by_thread_id: dict[tuple[str, int], Portal] = {}
  106. config: Config
  107. matrix: m.MatrixHandler
  108. az: AppService
  109. private_chat_portal_meta: bool
  110. _main_intent: IntentAPI | None
  111. _create_room_lock: asyncio.Lock
  112. backfill_lock: SimpleLock
  113. _msgid_dedup: deque[str]
  114. _reqid_dedup: set[str]
  115. _reaction_dedup: deque[tuple[str, int, str]]
  116. _last_participant_update: set[int]
  117. _reaction_lock: asyncio.Lock
  118. _backfill_leave: set[IntentAPI] | None
  119. _typing: set[UserID]
  120. def __init__(
  121. self,
  122. thread_id: str,
  123. receiver: int,
  124. other_user_pk: int | None,
  125. mxid: RoomID | None = None,
  126. name: str | None = None,
  127. avatar_url: ContentURI | None = None,
  128. encrypted: bool = False,
  129. name_set: bool = False,
  130. avatar_set: bool = False,
  131. relay_user_id: UserID | None = None,
  132. ) -> None:
  133. super().__init__(
  134. thread_id,
  135. receiver,
  136. other_user_pk,
  137. mxid,
  138. name,
  139. avatar_url,
  140. encrypted,
  141. name_set,
  142. avatar_set,
  143. relay_user_id,
  144. )
  145. self._create_room_lock = asyncio.Lock()
  146. self.log = self.log.getChild(thread_id)
  147. self._msgid_dedup = deque(maxlen=100)
  148. self._reaction_dedup = deque(maxlen=100)
  149. self._reqid_dedup = set()
  150. self._last_participant_update = set()
  151. self.backfill_lock = SimpleLock(
  152. "Waiting for backfilling to finish before handling %s", log=self.log
  153. )
  154. self._backfill_leave = None
  155. self._main_intent = None
  156. self._reaction_lock = asyncio.Lock()
  157. self._typing = set()
  158. self._relay_user = None
  159. @property
  160. def is_direct(self) -> bool:
  161. return self.other_user_pk is not None
  162. @property
  163. def main_intent(self) -> IntentAPI:
  164. if not self._main_intent:
  165. raise ValueError("Portal must be postinit()ed before main_intent can be used")
  166. return self._main_intent
  167. @classmethod
  168. def init_cls(cls, bridge: "InstagramBridge") -> None:
  169. cls.config = bridge.config
  170. cls.matrix = bridge.matrix
  171. cls.az = bridge.az
  172. cls.loop = bridge.loop
  173. cls.bridge = bridge
  174. cls.private_chat_portal_meta = cls.config["bridge.private_chat_portal_meta"]
  175. NotificationDisabler.puppet_cls = p.Puppet
  176. NotificationDisabler.config_enabled = cls.config["bridge.backfill.disable_notifications"]
  177. # region Misc
  178. async def _send_delivery_receipt(self, event_id: EventID) -> None:
  179. if event_id and self.config["bridge.delivery_receipts"]:
  180. try:
  181. await self.az.intent.mark_read(self.mxid, event_id)
  182. except Exception:
  183. self.log.exception("Failed to send delivery receipt for %s", event_id)
  184. async def _send_bridge_success(
  185. self,
  186. sender: u.User,
  187. event_id: EventID,
  188. event_type: EventType,
  189. msgtype: MessageType | None = None,
  190. ) -> None:
  191. sender.send_remote_checkpoint(
  192. status=MessageSendCheckpointStatus.SUCCESS,
  193. event_id=event_id,
  194. room_id=self.mxid,
  195. event_type=event_type,
  196. message_type=msgtype,
  197. )
  198. asyncio.create_task(self._send_message_status(event_id, err=None))
  199. await self._send_delivery_receipt(event_id)
  200. async def _send_bridge_error(
  201. self,
  202. sender: u.User,
  203. err: Exception,
  204. event_id: EventID,
  205. event_type: EventType,
  206. message_type: MessageType | None = None,
  207. confirmed: bool = False,
  208. ) -> None:
  209. sender.send_remote_checkpoint(
  210. self._status_from_exception(err),
  211. event_id,
  212. self.mxid,
  213. event_type,
  214. message_type=message_type,
  215. error=err,
  216. )
  217. if self.config["bridge.delivery_error_reports"]:
  218. event_type_str = {
  219. EventType.REACTION: "reaction",
  220. EventType.ROOM_REDACTION: "redaction",
  221. }.get(event_type, "message")
  222. error_type = "was not" if confirmed else "may not have been"
  223. await self._send_message(
  224. self.main_intent,
  225. TextMessageEventContent(
  226. msgtype=MessageType.NOTICE,
  227. body=f"\u26a0 Your {event_type_str} {error_type} bridged: {str(err)}",
  228. ),
  229. )
  230. asyncio.create_task(self._send_message_status(event_id, err))
  231. async def _send_message_status(self, event_id: EventID, err: Exception | None) -> None:
  232. if not self.config["bridge.message_status_events"]:
  233. return
  234. intent = self.az.intent if self.encrypted else self.main_intent
  235. status = BeeperMessageStatusEventContent(
  236. network=self.bridge_info_state_key,
  237. relates_to=RelatesTo(
  238. rel_type=RelationType.REFERENCE,
  239. event_id=event_id,
  240. ),
  241. success=err is None,
  242. )
  243. if err:
  244. status.reason = MessageStatusReason.GENERIC_ERROR
  245. status.error = str(err)
  246. status.is_certain = True
  247. status.can_retry = True
  248. if isinstance(err, NotImplementedError):
  249. status.can_retry = False
  250. status.reason = MessageStatusReason.UNSUPPORTED
  251. await intent.send_message_event(
  252. room_id=self.mxid,
  253. event_type=EventType.BEEPER_MESSAGE_STATUS,
  254. content=status,
  255. )
  256. async def _upsert_reaction(
  257. self,
  258. existing: DBReaction | None,
  259. intent: IntentAPI,
  260. mxid: EventID,
  261. message: DBMessage,
  262. sender: u.User | p.Puppet,
  263. reaction: str,
  264. mx_timestamp: int,
  265. ) -> None:
  266. if existing:
  267. self.log.debug(
  268. f"_upsert_reaction redacting {existing.mxid} and inserting {mxid}"
  269. f" (message: {message.mxid})"
  270. )
  271. await intent.redact(existing.mx_room, existing.mxid)
  272. await existing.edit(
  273. reaction=reaction, mxid=mxid, mx_room=message.mx_room, mx_timestamp=mx_timestamp
  274. )
  275. else:
  276. self.log.debug(f"_upsert_reaction inserting {mxid} (message: {message.mxid})")
  277. await DBReaction(
  278. mxid=mxid,
  279. mx_room=message.mx_room,
  280. ig_item_id=message.item_id,
  281. ig_receiver=self.receiver,
  282. ig_sender=sender.igpk,
  283. reaction=reaction,
  284. mx_timestamp=mx_timestamp,
  285. ).insert()
  286. # endregion
  287. # region Matrix event handling
  288. @staticmethod
  289. def _status_from_exception(e: Exception) -> MessageSendCheckpointStatus:
  290. if isinstance(e, NotImplementedError):
  291. return MessageSendCheckpointStatus.UNSUPPORTED
  292. return MessageSendCheckpointStatus.PERM_FAILURE
  293. async def handle_matrix_message(
  294. self, sender: u.User, message: MessageEventContent, event_id: EventID
  295. ) -> None:
  296. try:
  297. await self._handle_matrix_message(sender, message, event_id)
  298. except Exception as e:
  299. self.log.exception(f"Error handling Matrix event {event_id}")
  300. await self._send_bridge_error(
  301. sender,
  302. e,
  303. event_id,
  304. EventType.ROOM_MESSAGE,
  305. message_type=message.msgtype,
  306. confirmed=True,
  307. )
  308. else:
  309. await self._send_bridge_success(
  310. sender, event_id, EventType.ROOM_MESSAGE, message.msgtype
  311. )
  312. async def _handle_matrix_giphy(
  313. self,
  314. sender: u.User,
  315. event_id: EventID,
  316. request_id: str,
  317. giphy_id: str,
  318. ) -> CommandResponse:
  319. self.log.trace(f"Broadcasting giphy from {event_id} with request ID {request_id}")
  320. return await sender.client.broadcast(
  321. self.thread_id,
  322. ThreadItemType.ANIMATED_MEDIA,
  323. client_context=request_id,
  324. id=giphy_id,
  325. )
  326. async def _handle_matrix_image(
  327. self,
  328. sender: u.User,
  329. event_id: EventID,
  330. request_id: str,
  331. data: bytes,
  332. mime_type: str,
  333. width: int | None = None,
  334. height: int | None = None,
  335. ) -> CommandResponse:
  336. if mime_type != "image/jpeg":
  337. if Image is None:
  338. raise NotImplementedError(
  339. "Instagram does not allow non-JPEG images, and Pillow is not installed, "
  340. "so the bridge couldn't convert the image automatically"
  341. )
  342. with BytesIO(data) as inp, BytesIO() as out:
  343. img = Image.open(inp)
  344. img.convert("RGB").save(out, format="JPEG", quality=80)
  345. data = out.getvalue()
  346. mime_type = "image/jpeg"
  347. self.log.trace(f"Uploading photo from {event_id} (mime: {mime_type})")
  348. upload_resp = await sender.client.upload_photo(
  349. data, mime=mime_type, width=width, height=height
  350. )
  351. self.log.trace(f"Broadcasting uploaded photo with request ID {request_id}")
  352. return await sender.client.broadcast(
  353. self.thread_id,
  354. ThreadItemType.CONFIGURE_PHOTO,
  355. client_context=request_id,
  356. upload_id=upload_resp.upload_id,
  357. allow_full_aspect_ratio="true",
  358. )
  359. async def _handle_matrix_video(
  360. self,
  361. sender: u.User,
  362. event_id: EventID,
  363. request_id: str,
  364. data: bytes,
  365. mime_type: str,
  366. duration: int | None = None,
  367. width: int | None = None,
  368. height: int | None = None,
  369. ) -> CommandResponse:
  370. if mime_type != "video/mp4":
  371. data = await ffmpeg.convert_bytes(
  372. data,
  373. output_extension=".mp4",
  374. output_args=("-c:v", "libx264", "-c:a", "aac"),
  375. input_mime=mime_type,
  376. )
  377. self.log.trace(f"Uploading video from {event_id}")
  378. _, upload_id = await sender.client.upload_mp4(
  379. data, duration_ms=duration, width=width, height=height
  380. )
  381. self.log.trace(f"Broadcasting uploaded video with request ID {request_id}")
  382. return await sender.client.broadcast(
  383. self.thread_id,
  384. ThreadItemType.CONFIGURE_VIDEO,
  385. client_context=request_id,
  386. upload_id=upload_id,
  387. video_result="",
  388. )
  389. async def _handle_matrix_audio(
  390. self,
  391. sender: u.User,
  392. event_id: EventID,
  393. request_id: str,
  394. data: bytes,
  395. mime_type: str,
  396. waveform: list[int],
  397. duration: int | None = None,
  398. ) -> CommandResponse:
  399. if mime_type != "audio/mp4":
  400. data = await ffmpeg.convert_bytes(
  401. data, output_extension=".m4a", output_args=("-c:a", "aac"), input_mime=mime_type
  402. )
  403. self.log.trace(f"Uploading audio from {event_id}")
  404. _, upload_id = await sender.client.upload_mp4(data, audio=True, duration_ms=duration)
  405. self.log.trace(f"Broadcasting uploaded audio with request ID {request_id}")
  406. return await sender.client.broadcast_audio(
  407. self.thread_id,
  408. is_direct=self.is_direct,
  409. client_context=request_id,
  410. upload_id=upload_id,
  411. waveform=json.dumps([(part or 0) / 1024 for part in waveform], separators=(",", ":")),
  412. waveform_sampling_frequency_hz="10",
  413. )
  414. async def _handle_matrix_message(
  415. self, orig_sender: u.User, message: MessageEventContent, event_id: EventID
  416. ) -> None:
  417. sender, is_relay = await self.get_relay_sender(orig_sender, f"message {event_id}")
  418. assert sender, "user is not logged in"
  419. assert sender.is_connected, "You're not connected to Instagram"
  420. if is_relay:
  421. await self.apply_relay_message_format(orig_sender, message)
  422. reply_to = {}
  423. if message.get_reply_to():
  424. msg = await DBMessage.get_by_mxid(message.get_reply_to(), self.mxid)
  425. if msg and msg.client_context:
  426. reply_to = {
  427. "replied_to_item_id": msg.item_id,
  428. "replied_to_client_context": msg.client_context,
  429. }
  430. request_id = sender.state.gen_client_context()
  431. self._reqid_dedup.add(request_id)
  432. self.log.debug(
  433. f"Handling Matrix message {event_id} from {sender.mxid}/{sender.igpk} "
  434. f"with request ID {request_id}"
  435. )
  436. if message.msgtype in (MessageType.EMOTE, MessageType.TEXT, MessageType.NOTICE):
  437. text = message.body
  438. if message.msgtype == MessageType.EMOTE:
  439. text = f"/me {text}"
  440. self.log.trace(f"Sending Matrix text from {event_id} with request ID {request_id}")
  441. urls = SIMPLE_URL_REGEX.findall(text) or None
  442. if not self.is_direct:
  443. # Instagram groups don't seem to support sending link previews,
  444. # and the client_context-based deduplication breaks when trying to send them.
  445. urls = None
  446. resp = await sender.mqtt.send_text(
  447. self.thread_id, text=text, urls=urls, client_context=request_id, **reply_to
  448. )
  449. elif message.msgtype.is_media and "fi.mau.instagram.giphy_id" in message:
  450. resp = await self._handle_matrix_giphy(
  451. sender, event_id, request_id, message["fi.mau.instagram.giphy_id"]
  452. )
  453. elif message.msgtype.is_media:
  454. if message.file and decrypt_attachment:
  455. data = await self.main_intent.download_media(message.file.url)
  456. data = decrypt_attachment(
  457. data, message.file.key.key, message.file.hashes.get("sha256"), message.file.iv
  458. )
  459. else:
  460. data = await self.main_intent.download_media(message.url)
  461. mime_type = message.info.mimetype or magic.from_buffer(data, mime=True)
  462. if message.msgtype == MessageType.IMAGE:
  463. resp = await self._handle_matrix_image(
  464. sender,
  465. event_id,
  466. request_id,
  467. data,
  468. mime_type,
  469. width=message.info.width,
  470. height=message.info.height,
  471. )
  472. elif message.msgtype == MessageType.AUDIO:
  473. waveform = message.get("org.matrix.msc1767.audio", {}).get("waveform", [0] * 30)
  474. resp = await self._handle_matrix_audio(
  475. sender,
  476. event_id,
  477. request_id,
  478. data,
  479. mime_type,
  480. waveform,
  481. duration=message.info.duration,
  482. )
  483. elif message.msgtype == MessageType.VIDEO:
  484. resp = await self._handle_matrix_video(
  485. sender,
  486. event_id,
  487. request_id,
  488. data,
  489. mime_type,
  490. duration=message.info.duration,
  491. width=message.info.width,
  492. height=message.info.height,
  493. )
  494. else:
  495. raise NotImplementedError(
  496. "Non-image/video/audio files are currently not supported"
  497. )
  498. else:
  499. raise NotImplementedError(f"Unknown message type {message.msgtype}")
  500. self.log.trace(f"Got response to message send {request_id}: {resp}")
  501. if resp.status != "ok":
  502. self.log.warning(f"Failed to handle {event_id}: {resp}")
  503. raise Exception(f"Failed to handle event. Error: {resp.payload.message}")
  504. else:
  505. self._msgid_dedup.appendleft(resp.payload.item_id)
  506. try:
  507. await DBMessage(
  508. mxid=event_id,
  509. mx_room=self.mxid,
  510. item_id=resp.payload.item_id,
  511. client_context=resp.payload.client_context,
  512. receiver=self.receiver,
  513. sender=sender.igpk,
  514. ig_timestamp=int(resp.payload.timestamp),
  515. ).insert()
  516. except asyncpg.UniqueViolationError as e:
  517. self.log.warning(
  518. f"Error while persisting {event_id} ({resp.payload.client_context}) "
  519. f"-> {resp.payload.item_id}: {e}"
  520. )
  521. self._reqid_dedup.remove(request_id)
  522. self.log.debug(
  523. f"Handled Matrix message {event_id} ({resp.payload.client_context}) "
  524. f"-> {resp.payload.item_id}"
  525. )
  526. async def handle_matrix_reaction(
  527. self, sender: u.User, event_id: EventID, reacting_to: EventID, emoji: str, timestamp: int
  528. ) -> None:
  529. try:
  530. await self._handle_matrix_reaction(sender, event_id, reacting_to, emoji, timestamp)
  531. except Exception as e:
  532. self.log.exception(f"Error handling Matrix reaction {event_id}")
  533. await self._send_bridge_error(
  534. sender,
  535. e,
  536. event_id,
  537. EventType.REACTION,
  538. confirmed=True,
  539. )
  540. else:
  541. await self._send_bridge_success(sender, event_id, EventType.REACTION)
  542. async def _handle_matrix_reaction(
  543. self, sender: u.User, event_id: EventID, reacting_to: EventID, emoji: str, timestamp: int
  544. ) -> None:
  545. if not await sender.is_logged_in():
  546. self.log.debug(f"Ignoring reaction by non-logged-in user {sender.mxid}")
  547. raise NotImplementedError("User is not logged in")
  548. message = await DBMessage.get_by_mxid(reacting_to, self.mxid)
  549. if not message or message.is_internal:
  550. self.log.debug(f"Ignoring reaction to unknown event {reacting_to}")
  551. await self.main_intent.redact(self.mxid, event_id, reason="Unknown target message")
  552. raise NotImplementedError("Unknown target message")
  553. existing = await DBReaction.get_by_item_id(message.item_id, message.receiver, sender.igpk)
  554. if existing and existing.reaction == emoji:
  555. return
  556. dedup_id = (message.item_id, sender.igpk, emoji)
  557. self._reaction_dedup.appendleft(dedup_id)
  558. async with self._reaction_lock:
  559. resp = await sender.mqtt.send_reaction(
  560. self.thread_id, item_id=message.item_id, emoji=emoji
  561. )
  562. if resp.status != "ok":
  563. if resp.payload.message == "invalid unicode emoji":
  564. # Instagram doesn't support this reaction. Notify the user, and redact it
  565. # so that it doesn't get confusing.
  566. await self.main_intent.redact(self.mxid, event_id, reason="Unsupported emoji")
  567. raise NotImplementedError(f"Instagram does not support the {emoji} emoji.")
  568. raise Exception(f"Unknown response error: {resp}")
  569. self.log.trace(f"{sender.mxid} reacted to {message.item_id} with {emoji}")
  570. await self._upsert_reaction(
  571. existing, self.main_intent, event_id, message, sender, emoji, timestamp
  572. )
  573. async def handle_matrix_redaction(
  574. self, orig_sender: u.User, event_id: EventID, redaction_event_id: EventID
  575. ) -> None:
  576. sender = None
  577. try:
  578. sender, _ = await self.get_relay_sender(orig_sender, f"redaction {event_id}")
  579. if not sender:
  580. raise Exception("User is not logged in")
  581. await self._handle_matrix_redaction(sender, event_id)
  582. except Exception as e:
  583. self.log.exception(f"Error handling Matrix redaction {event_id}")
  584. await self._send_bridge_error(
  585. sender or orig_sender,
  586. e,
  587. redaction_event_id,
  588. EventType.ROOM_REDACTION,
  589. confirmed=True,
  590. )
  591. else:
  592. await self._send_bridge_success(sender, redaction_event_id, EventType.ROOM_REDACTION)
  593. async def _handle_matrix_redaction(self, sender: u.User, event_id: EventID) -> None:
  594. if not sender.is_connected:
  595. raise Exception("You're not connected to Instagram")
  596. reaction = await DBReaction.get_by_mxid(event_id, self.mxid)
  597. if reaction:
  598. try:
  599. await reaction.delete()
  600. await sender.mqtt.send_reaction(
  601. self.thread_id,
  602. item_id=reaction.ig_item_id,
  603. reaction_status=ReactionStatus.DELETED,
  604. emoji="",
  605. )
  606. except Exception as e:
  607. raise Exception(f"Removing reaction failed: {e}")
  608. else:
  609. self.log.trace(f"Removed reaction to {reaction.ig_item_id} after Matrix redaction")
  610. return
  611. message = await DBMessage.get_by_mxid(event_id, self.mxid)
  612. if message and not message.is_internal:
  613. try:
  614. await message.delete()
  615. await sender.client.delete_item(self.thread_id, message.item_id)
  616. self.log.trace(f"Removed {message} after Matrix redaction")
  617. except Exception as e:
  618. raise Exception(f"Removing message failed: {e}")
  619. else:
  620. self.log.trace(f"Removed message {message.item_id} after Matrix redaction")
  621. return
  622. raise NotImplementedError("No message or reaction found for redaction")
  623. async def handle_matrix_typing(self, users: set[UserID]) -> None:
  624. if users == self._typing:
  625. return
  626. old_typing = self._typing
  627. self._typing = users
  628. await self._handle_matrix_typing(old_typing - users, TypingStatus.OFF)
  629. await self._handle_matrix_typing(users - old_typing, TypingStatus.TEXT)
  630. async def _handle_matrix_typing(self, users: set[UserID], status: TypingStatus) -> None:
  631. for mxid in users:
  632. user = await u.User.get_by_mxid(mxid, create=False)
  633. if (
  634. not user
  635. or not await user.is_logged_in()
  636. or user.remote_typing_status == status
  637. or not user.is_connected
  638. ):
  639. continue
  640. user.remote_typing_status = None
  641. await user.mqtt.indicate_activity(self.thread_id, status)
  642. async def handle_matrix_leave(self, user: u.User) -> None:
  643. if not await user.is_logged_in():
  644. return
  645. if self.is_direct:
  646. self.log.info(f"{user.mxid} left private chat portal with {self.other_user_pk}")
  647. if user.igpk == self.receiver:
  648. self.log.info(
  649. f"{user.mxid} was the recipient of this portal. Cleaning up and deleting..."
  650. )
  651. await self.cleanup_and_delete()
  652. else:
  653. self.log.debug(f"{user.mxid} left portal to {self.thread_id}")
  654. # TODO cleanup if empty
  655. # endregion
  656. # region Instagram event handling
  657. async def _reupload_instagram_media(
  658. self, source: u.User, media: RegularMediaItem, intent: IntentAPI
  659. ) -> MediaMessageEventContent:
  660. if media.media_type == MediaType.IMAGE:
  661. image = media.best_image
  662. if not image:
  663. raise ValueError("Attachment not available: didn't find photo URL")
  664. url = image.url
  665. msgtype = MessageType.IMAGE
  666. info = ImageInfo(height=image.height, width=image.width)
  667. elif media.media_type == MediaType.VIDEO:
  668. video = media.best_video
  669. if not video:
  670. raise ValueError("Attachment not available: didn't find video URL")
  671. url = video.url
  672. msgtype = MessageType.VIDEO
  673. info = VideoInfo(height=video.height, width=video.width)
  674. elif media.media_type == MediaType.CAROUSEL:
  675. raise ValueError(
  676. "Carousel media is not currently supported, "
  677. "please view the post on Instagram via the link below"
  678. )
  679. else:
  680. raise ValueError(
  681. f"Attachment not available: unsupported media type {media.media_type.human_name}"
  682. )
  683. return await self._reupload_instagram_file(source, url, msgtype, info, intent)
  684. async def _reupload_instagram_animated(
  685. self, source: u.User, media: AnimatedMediaItem, intent: IntentAPI
  686. ) -> MediaMessageEventContent:
  687. url = media.images.fixed_height.webp
  688. info = ImageInfo(
  689. height=int(media.images.fixed_height.height),
  690. width=int(media.images.fixed_height.width),
  691. )
  692. return await self._reupload_instagram_file(source, url, MessageType.IMAGE, info, intent)
  693. async def _reupload_instagram_voice(
  694. self, source: u.User, media: VoiceMediaItem, intent: IntentAPI
  695. ) -> MediaMessageEventContent:
  696. async def convert_to_ogg(data, mimetype):
  697. converted = await ffmpeg.convert_bytes(
  698. data, ".ogg", output_args=("-c:a", "libopus"), input_mime=mimetype
  699. )
  700. return converted, "audio/ogg"
  701. url = media.media.audio.audio_src
  702. info = AudioInfo(duration=media.media.audio.duration)
  703. waveform = [int(p * 1024) for p in media.media.audio.waveform_data]
  704. content = await self._reupload_instagram_file(
  705. source, url, MessageType.AUDIO, info, intent, convert_to_ogg
  706. )
  707. content["org.matrix.msc1767.audio"] = {
  708. "duration": media.media.audio.duration,
  709. "waveform": waveform,
  710. }
  711. content["org.matrix.msc3245.voice"] = {}
  712. return content
  713. async def _reupload_instagram_file(
  714. self,
  715. source: u.User,
  716. url: str,
  717. msgtype: MessageType | None,
  718. info: ImageInfo | VideoInfo | AudioInfo,
  719. intent: IntentAPI,
  720. convert_fn: Callable[[bytes, str], Awaitable[tuple[bytes, str]]] | None = None,
  721. ) -> MediaMessageEventContent:
  722. async with source.client.raw_http_get(url) as resp:
  723. try:
  724. length = int(resp.headers["Content-Length"])
  725. except KeyError:
  726. # TODO can the download be short-circuited if there's too much data?
  727. self.log.warning(
  728. "Got file download response with no Content-Length header,"
  729. "reading data dangerously"
  730. )
  731. length = 0
  732. if length > self.matrix.media_config.upload_size:
  733. self.log.debug(
  734. f"{url} was too large ({length} > {self.matrix.media_config.upload_size})"
  735. )
  736. raise ValueError("Attachment not available: too large")
  737. data = await resp.read()
  738. info.mimetype = resp.headers["Content-Type"] or magic.from_buffer(data, mime=True)
  739. # Run the conversion function on the data.
  740. if convert_fn is not None:
  741. data, info.mimetype = await convert_fn(data, info.mimetype)
  742. info.size = len(data)
  743. extension = {
  744. "image/webp": ".webp",
  745. "image/jpeg": ".jpg",
  746. "video/mp4": ".mp4",
  747. "audio/mp4": ".m4a",
  748. "audio/ogg": ".ogg",
  749. }.get(info.mimetype)
  750. extension = extension or mimetypes.guess_extension(info.mimetype) or ""
  751. file_name = f"{msgtype.value[2:]}{extension}" if msgtype else None
  752. upload_mime_type = info.mimetype
  753. upload_file_name = file_name
  754. decryption_info = None
  755. if self.encrypted and encrypt_attachment:
  756. data, decryption_info = encrypt_attachment(data)
  757. upload_mime_type = "application/octet-stream"
  758. upload_file_name = None
  759. mxc = await intent.upload_media(
  760. data,
  761. mime_type=upload_mime_type,
  762. filename=upload_file_name,
  763. async_upload=self.config["homeserver.async_media"],
  764. )
  765. if decryption_info:
  766. decryption_info.url = mxc
  767. mxc = None
  768. return MediaMessageEventContent(
  769. body=file_name,
  770. external_url=url,
  771. url=mxc,
  772. file=decryption_info,
  773. info=info,
  774. msgtype=msgtype,
  775. )
  776. def _get_instagram_media_info(self, item: ThreadItem) -> tuple[MediaUploadFunc, MediaData]:
  777. # TODO maybe use a dict and item.item_type instead of a ton of ifs
  778. method = self._reupload_instagram_media
  779. if item.media:
  780. media_data = item.media
  781. elif item.visual_media:
  782. media_data = item.visual_media.media
  783. elif item.animated_media:
  784. media_data = item.animated_media
  785. method = self._reupload_instagram_animated
  786. elif item.voice_media:
  787. media_data = item.voice_media
  788. method = self._reupload_instagram_voice
  789. elif item.reel_share:
  790. media_data = item.reel_share.media
  791. elif item.story_share:
  792. media_data = item.story_share.media
  793. elif item.clip:
  794. media_data = item.clip.clip
  795. elif item.felix_share and item.felix_share.video:
  796. media_data = item.felix_share.video
  797. elif item.media_share:
  798. media_data = item.media_share
  799. elif item.direct_media_share:
  800. media_data = item.direct_media_share.media
  801. else:
  802. self.log.debug(f"Unknown media type in {item}")
  803. raise ValueError("Attachment not available: unsupported media type")
  804. if not media_data:
  805. self.log.debug(f"Didn't get media_data in {item}")
  806. raise ValueError("Attachment not available: unsupported media type")
  807. elif isinstance(media_data, ExpiredMediaItem):
  808. self.log.debug(f"Expired media in item {item}")
  809. raise ValueError("Attachment not available: media expired")
  810. return method, media_data
  811. async def _handle_instagram_media(
  812. self, source: u.User, intent: IntentAPI, item: ThreadItem
  813. ) -> EventID | None:
  814. try:
  815. reupload_func, media_data = self._get_instagram_media_info(item)
  816. content = await reupload_func(source, media_data, intent)
  817. except ValueError as e:
  818. content = TextMessageEventContent(body=str(e), msgtype=MessageType.NOTICE)
  819. except Exception:
  820. self.log.warning("Failed to upload media", exc_info=True)
  821. content = TextMessageEventContent(
  822. body="Attachment not available: failed to copy file", msgtype=MessageType.NOTICE
  823. )
  824. await self._add_instagram_reply(content, item.replied_to_message)
  825. return await self._send_message(intent, content, timestamp=item.timestamp_ms)
  826. async def _handle_instagram_media_share(
  827. self, source: u.User, intent: IntentAPI, item: ThreadItem
  828. ) -> EventID | None:
  829. item_type_name = None
  830. if item.media_share:
  831. share_item = item.media_share
  832. elif item.clip:
  833. share_item = item.clip.clip
  834. item_type_name = "clip"
  835. elif item.felix_share and item.felix_share.video:
  836. share_item = item.felix_share.video
  837. elif item.story_share:
  838. share_item = item.story_share.media
  839. item_type_name = "story"
  840. elif item.direct_media_share:
  841. share_item = item.direct_media_share.media
  842. else:
  843. return None
  844. item_type_name = item_type_name or share_item.media_type.human_name
  845. user_text = f"@{share_item.user.username}"
  846. user_link = (
  847. f'<a href="https://www.instagram.com/{share_item.user.username}/">{user_text}</a>'
  848. )
  849. prefix = TextMessageEventContent(
  850. msgtype=MessageType.NOTICE,
  851. format=Format.HTML,
  852. body=f"Sent {user_text}'s {item_type_name}",
  853. formatted_body=f"Sent {user_link}'s {item_type_name}",
  854. )
  855. if item.direct_media_share and item.direct_media_share.media_share_type == "tag":
  856. tagged_user_id = item.direct_media_share.tagged_user_id
  857. if tagged_user_id == source.igpk and share_item.user.pk == self.other_user_pk:
  858. prefix.body = prefix.formatted_body = "Tagged you in their post"
  859. elif share_item.user.pk == source.igpk and tagged_user_id == self.other_user_pk:
  860. prefix.body = prefix.formatted_body = "Tagged them in your post"
  861. await self._send_message(intent, prefix, timestamp=item.timestamp_ms)
  862. event_id = await self._handle_instagram_media(source, intent, item)
  863. external_url = f"https://www.instagram.com/p/{share_item.code}/"
  864. if share_item.caption:
  865. caption_body = (
  866. f"> {share_item.caption.user.username}: {share_item.caption.text}\n\n"
  867. f"{external_url}"
  868. )
  869. caption_formatted_body = (
  870. f"<blockquote><strong>{share_item.caption.user.username}</strong>"
  871. f" {share_item.caption.text}</blockquote>"
  872. f'<a href="{external_url}">instagram.com/p/{share_item.code}</a>'
  873. )
  874. else:
  875. caption_body = external_url
  876. caption_formatted_body = (
  877. f'<a href="{external_url}">instagram.com/p/{share_item.code}</a>'
  878. )
  879. caption = TextMessageEventContent(
  880. msgtype=MessageType.TEXT,
  881. body=caption_body,
  882. formatted_body=caption_formatted_body,
  883. format=Format.HTML,
  884. external_url=external_url,
  885. )
  886. await self._send_message(intent, caption, timestamp=item.timestamp_ms)
  887. return event_id
  888. async def _handle_instagram_reel_share(
  889. self, source: u.User, intent: IntentAPI, item: ThreadItem
  890. ) -> EventID | None:
  891. media = item.reel_share.media
  892. prefix_html = None
  893. if item.reel_share.type == ReelShareType.REPLY:
  894. if item.reel_share.reel_owner_id == source.igpk:
  895. prefix = "Replied to your story"
  896. else:
  897. username = media.user.username
  898. prefix = f"Sent @{username}'s story"
  899. user_link = f'<a href="https://www.instagram.com/{username}/">@{username}</a>'
  900. prefix_html = f"Sent {user_link}'s story"
  901. elif item.reel_share.type == ReelShareType.REACTION:
  902. if item.reel_share.reel_owner_id == source.igpk:
  903. prefix = "Reacted to your story"
  904. elif item.user_id == source.igpk:
  905. prefix = "You reacted to their story"
  906. else:
  907. prefix = "Reacted to a story"
  908. elif item.reel_share.type == ReelShareType.MENTION:
  909. if item.reel_share.mentioned_user_id == source.igpk:
  910. prefix = "Mentioned you in their story"
  911. else:
  912. prefix = "You mentioned them in your story"
  913. else:
  914. self.log.debug(f"Unsupported reel share type {item.reel_share.type}")
  915. return None
  916. prefix_content = TextMessageEventContent(msgtype=MessageType.NOTICE, body=prefix)
  917. if prefix_html:
  918. prefix_content.format = Format.HTML
  919. prefix_content.formatted_body = prefix_html
  920. content = TextMessageEventContent(msgtype=MessageType.TEXT, body=item.reel_share.text)
  921. if not content.body and isinstance(media, MediaShareItem):
  922. content.body = media.caption.text if media.caption else ""
  923. if not content.body:
  924. content.body = "<no caption>"
  925. await self._send_message(intent, prefix_content, timestamp=item.timestamp_ms)
  926. if isinstance(media, ExpiredMediaItem):
  927. # TODO send message about expired story
  928. pass
  929. else:
  930. fake_item_id = f"fi.mau.instagram.reel_share.{item.user_id}.{media.pk}"
  931. existing = await DBMessage.get_by_item_id(fake_item_id, self.receiver)
  932. if existing:
  933. # If the user already reacted or replied to the same reel share item,
  934. # use a Matrix reply instead of reposting the image.
  935. content.set_reply(existing.mxid)
  936. else:
  937. media_event_id = await self._handle_instagram_media(source, intent, item)
  938. await DBMessage(
  939. mxid=media_event_id,
  940. mx_room=self.mxid,
  941. item_id=fake_item_id,
  942. client_context=None,
  943. receiver=self.receiver,
  944. sender=media.user.pk,
  945. ig_timestamp=None,
  946. ).insert()
  947. return await self._send_message(intent, content, timestamp=item.timestamp_ms)
  948. async def _handle_instagram_link(
  949. self,
  950. source: u.User,
  951. intent: IntentAPI,
  952. item: ThreadItem,
  953. ) -> EventID:
  954. content = TextMessageEventContent(msgtype=MessageType.TEXT, body=item.link.text)
  955. link = item.link.link_context
  956. preview = {
  957. "og:url": link.link_url,
  958. "og:title": link.link_title,
  959. "og:description": link.link_summary,
  960. }
  961. if link.link_image_url:
  962. reuploaded = await self._reupload_instagram_file(
  963. source, link.link_image_url, msgtype=None, info=ImageInfo(), intent=intent
  964. )
  965. preview["og:image"] = reuploaded.url
  966. preview["og:image:type"] = reuploaded.info.mimetype
  967. preview["og:image:width"] = reuploaded.info.width
  968. preview["og:image:height"] = reuploaded.info.height
  969. preview["matrix:image:size"] = reuploaded.info.size
  970. if reuploaded.file:
  971. preview["beeper:image:encryption"] = reuploaded.file.serialize()
  972. preview = {k: v for k, v in preview.items() if v}
  973. content["com.beeper.linkpreviews"] = [preview] if "og:title" in preview else []
  974. await self._add_instagram_reply(content, item.replied_to_message)
  975. return await self._send_message(intent, content, timestamp=item.timestamp_ms)
  976. async def _handle_instagram_text(
  977. self, intent: IntentAPI, item: ThreadItem, text: str
  978. ) -> EventID:
  979. content = TextMessageEventContent(msgtype=MessageType.TEXT, body=text)
  980. content["com.beeper.linkpreviews"] = []
  981. await self._add_instagram_reply(content, item.replied_to_message)
  982. return await self._send_message(intent, content, timestamp=item.timestamp_ms)
  983. async def _send_instagram_unhandled(self, intent: IntentAPI, item: ThreadItem) -> EventID:
  984. content = TextMessageEventContent(
  985. msgtype=MessageType.NOTICE, body=f"Unsupported message type {item.item_type.value}"
  986. )
  987. await self._add_instagram_reply(content, item.replied_to_message)
  988. return await self._send_message(intent, content, timestamp=item.timestamp_ms)
  989. async def _handle_instagram_location(
  990. self, intent: IntentAPI, item: ThreadItem
  991. ) -> EventID | None:
  992. loc = item.location
  993. if not loc.lng or not loc.lat:
  994. # TODO handle somehow
  995. return None
  996. long_char = "E" if loc.lng > 0 else "W"
  997. lat_char = "N" if loc.lat > 0 else "S"
  998. body = (
  999. f"{loc.name} - {round(abs(loc.lat), 4)}° {lat_char}, "
  1000. f"{round(abs(loc.lng), 4)}° {long_char}"
  1001. )
  1002. url = f"https://www.openstreetmap.org/#map=15/{loc.lat}/{loc.lng}"
  1003. external_url = None
  1004. if loc.external_source == "facebook_places":
  1005. external_url = f"https://www.facebook.com/{loc.short_name}-{loc.facebook_places_id}"
  1006. content = LocationMessageEventContent(
  1007. msgtype=MessageType.LOCATION,
  1008. geo_uri=f"geo:{loc.lat},{loc.lng}",
  1009. body=f"Location: {body}\n{url}",
  1010. external_url=external_url,
  1011. )
  1012. content["format"] = str(Format.HTML)
  1013. content["formatted_body"] = f"Location: <a href='{url}'>{body}</a>"
  1014. await self._add_instagram_reply(content, item.replied_to_message)
  1015. return await self._send_message(intent, content, timestamp=item.timestamp_ms)
  1016. async def _handle_instagram_profile(
  1017. self, intent: IntentAPI, item: ThreadItem
  1018. ) -> EventID | None:
  1019. username = item.profile.username
  1020. user_link = f'<a href="https://www.instagram.com/{username}/">@{username}</a>'
  1021. text = f"Shared @{username}'s profile"
  1022. html = f"Shared {user_link}'s profile"
  1023. content = TextMessageEventContent(
  1024. msgtype=MessageType.TEXT, format=Format.HTML, body=text, formatted_body=html
  1025. )
  1026. await self._add_instagram_reply(content, item.replied_to_message)
  1027. return await self._send_message(intent, content, timestamp=item.timestamp_ms)
  1028. async def handle_instagram_item(
  1029. self, source: u.User, sender: p.Puppet, item: ThreadItem, is_backfill: bool = False
  1030. ) -> None:
  1031. try:
  1032. await self._handle_instagram_item(source, sender, item, is_backfill)
  1033. except Exception:
  1034. self.log.exception("Fatal error handling Instagram item")
  1035. self.log.trace("Item content: %s", item.serialize())
  1036. async def _add_instagram_reply(
  1037. self, content: MessageEventContent, reply_to: ThreadItem | None
  1038. ) -> None:
  1039. if not reply_to:
  1040. return
  1041. message = await DBMessage.get_by_item_id(reply_to.item_id, self.receiver)
  1042. if not message:
  1043. return
  1044. content.set_reply(message.mxid)
  1045. if not isinstance(content, TextMessageEventContent):
  1046. return
  1047. try:
  1048. evt = await self.main_intent.get_event(message.mx_room, message.mxid)
  1049. except (MNotFound, MForbidden):
  1050. evt = None
  1051. if not evt:
  1052. return
  1053. if evt.type == EventType.ROOM_ENCRYPTED:
  1054. try:
  1055. evt = await self.matrix.e2ee.decrypt(evt, wait_session_timeout=0)
  1056. except SessionNotFound:
  1057. return
  1058. if isinstance(evt.content, TextMessageEventContent):
  1059. evt.content.trim_reply_fallback()
  1060. content.set_reply(evt)
  1061. async def _handle_instagram_item(
  1062. self, source: u.User, sender: p.Puppet, item: ThreadItem, is_backfill: bool = False
  1063. ) -> None:
  1064. if not isinstance(item, ThreadItem):
  1065. # Parsing these items failed, they should have been logged already
  1066. return
  1067. client_context = item.client_context
  1068. link_client_context = item.link.client_context if item.link else None
  1069. cc = client_context
  1070. if link_client_context:
  1071. if not client_context:
  1072. cc = f"link:{link_client_context}"
  1073. elif client_context != link_client_context:
  1074. cc = f"{client_context}/link:{link_client_context}"
  1075. if client_context and client_context in self._reqid_dedup:
  1076. self.log.debug(
  1077. f"Ignoring message {item.item_id} ({cc}) by {item.user_id}"
  1078. " as it was sent by us (client_context in dedup queue)"
  1079. )
  1080. return
  1081. elif link_client_context and link_client_context in self._reqid_dedup:
  1082. self.log.debug(
  1083. f"Ignoring message {item.item_id} ({cc}) by {item.user_id}"
  1084. " as it was sent by us (link.client_context in dedup queue)"
  1085. )
  1086. return
  1087. if item.item_id in self._msgid_dedup:
  1088. self.log.debug(
  1089. f"Ignoring message {item.item_id} ({cc}) by {item.user_id}"
  1090. " as it was already handled (message.id in dedup queue)"
  1091. )
  1092. return
  1093. self._msgid_dedup.appendleft(item.item_id)
  1094. if await DBMessage.get_by_item_id(item.item_id, self.receiver) is not None:
  1095. self.log.debug(
  1096. f"Ignoring message {item.item_id} ({cc}) by {item.user_id}"
  1097. " as it was already handled (message.id in database)"
  1098. )
  1099. return
  1100. self.log.debug(f"Starting handling of message {item.item_id} ({cc}) by {item.user_id}")
  1101. asyncio.create_task(sender.intent_for(self).set_typing(self.mxid, is_typing=False))
  1102. await self._handle_deduplicated_instagram_item(source, sender, item, is_backfill)
  1103. async def _handle_deduplicated_instagram_item(
  1104. self, source: u.User, sender: p.Puppet, item: ThreadItem, is_backfill: bool = False
  1105. ) -> None:
  1106. if self.backfill_lock.locked and sender.need_backfill_invite(self):
  1107. self.log.debug("Adding %s's default puppet to room for backfilling", sender.mxid)
  1108. if self.is_direct:
  1109. await self.main_intent.invite_user(self.mxid, sender.default_mxid)
  1110. intent = sender.default_mxid_intent
  1111. await intent.ensure_joined(self.mxid)
  1112. self._backfill_leave.add(intent)
  1113. else:
  1114. intent = sender.intent_for(self)
  1115. event_id = None
  1116. needs_handling = True
  1117. if item.media or item.animated_media or item.voice_media or item.visual_media:
  1118. event_id = await self._handle_instagram_media(source, intent, item)
  1119. elif item.location:
  1120. event_id = await self._handle_instagram_location(intent, item)
  1121. elif item.profile:
  1122. event_id = await self._handle_instagram_profile(intent, item)
  1123. elif item.reel_share:
  1124. event_id = await self._handle_instagram_reel_share(source, intent, item)
  1125. elif (
  1126. item.media_share
  1127. or item.direct_media_share
  1128. or item.story_share
  1129. or item.clip
  1130. or item.felix_share
  1131. ):
  1132. event_id = await self._handle_instagram_media_share(source, intent, item)
  1133. elif item.action_log:
  1134. # These probably don't need to be bridged
  1135. needs_handling = False
  1136. self.log.debug(f"Ignoring action log message {item.item_id}")
  1137. # TODO handle item.clip?
  1138. if item.text:
  1139. event_id = await self._handle_instagram_text(intent, item, item.text)
  1140. elif item.like:
  1141. # We handle likes as text because Matrix clients do big emoji on their own.
  1142. event_id = await self._handle_instagram_text(intent, item, item.like)
  1143. elif item.link:
  1144. event_id = await self._handle_instagram_link(source, intent, item)
  1145. handled = bool(event_id)
  1146. if not event_id and needs_handling:
  1147. self.log.debug(f"Unhandled Instagram message {item.item_id}")
  1148. event_id = await self._send_instagram_unhandled(intent, item)
  1149. cc = item.client_context
  1150. if not cc and item.link and item.link.client_context:
  1151. cc = item.link.client_context
  1152. msg = DBMessage(
  1153. mxid=event_id,
  1154. mx_room=self.mxid,
  1155. item_id=item.item_id,
  1156. client_context=cc,
  1157. receiver=self.receiver,
  1158. sender=sender.pk,
  1159. ig_timestamp=item.timestamp,
  1160. )
  1161. await msg.insert()
  1162. await self._send_delivery_receipt(event_id)
  1163. if handled:
  1164. self.log.debug(f"Handled Instagram message {item.item_id} -> {event_id}")
  1165. elif needs_handling:
  1166. self.log.debug(
  1167. f"Unhandled Instagram message {item.item_id} "
  1168. f"(type {item.item_type} -> fallback error {event_id})"
  1169. )
  1170. if is_backfill and item.reactions:
  1171. await self._handle_instagram_reactions(msg, item.reactions.emojis, is_backfill=True)
  1172. async def handle_instagram_remove(self, item_id: str) -> None:
  1173. message = await DBMessage.get_by_item_id(item_id, self.receiver)
  1174. if message is None:
  1175. return
  1176. await message.delete()
  1177. if message.mxid:
  1178. sender = await p.Puppet.get_by_pk(message.sender)
  1179. try:
  1180. await sender.intent_for(self).redact(self.mxid, message.mxid)
  1181. except MForbidden:
  1182. await self.main_intent.redact(self.mxid, message.mxid)
  1183. self.log.debug(f"Redacted {message.mxid} after Instagram unsend")
  1184. async def _handle_instagram_reactions(
  1185. self, message: DBMessage, reactions: list[Reaction], is_backfill: bool = False
  1186. ) -> None:
  1187. old_reactions: dict[int, DBReaction]
  1188. old_reactions = {
  1189. reaction.ig_sender: reaction
  1190. for reaction in await DBReaction.get_all_by_item_id(message.item_id, self.receiver)
  1191. }
  1192. timestamp_deduplicator = 1
  1193. for new_reaction in reactions:
  1194. old_reaction = old_reactions.pop(new_reaction.sender_id, None)
  1195. if old_reaction and old_reaction.reaction == new_reaction.emoji:
  1196. continue
  1197. puppet = await p.Puppet.get_by_pk(new_reaction.sender_id)
  1198. intent = puppet.intent_for(self)
  1199. timestamp = new_reaction.timestamp_ms if is_backfill else int(time.time() * 1000)
  1200. if is_backfill:
  1201. timestamp += timestamp_deduplicator
  1202. timestamp_deduplicator += 1
  1203. reaction_event_id = await intent.react(
  1204. self.mxid, message.mxid, new_reaction.emoji, timestamp=timestamp
  1205. )
  1206. await self._upsert_reaction(
  1207. old_reaction,
  1208. intent,
  1209. reaction_event_id,
  1210. message,
  1211. puppet,
  1212. new_reaction.emoji,
  1213. timestamp,
  1214. )
  1215. for old_reaction in old_reactions.values():
  1216. await old_reaction.delete()
  1217. puppet = await p.Puppet.get_by_pk(old_reaction.ig_sender)
  1218. await puppet.intent_for(self).redact(self.mxid, old_reaction.mxid)
  1219. async def handle_instagram_update(self, item: MessageSyncMessage) -> None:
  1220. message = await DBMessage.get_by_item_id(item.item_id, self.receiver)
  1221. if not message:
  1222. return
  1223. if item.has_seen:
  1224. puppet = await p.Puppet.get_by_pk(item.has_seen, create=False)
  1225. if puppet:
  1226. await puppet.intent_for(self).mark_read(self.mxid, message.mxid)
  1227. else:
  1228. async with self._reaction_lock:
  1229. await self._handle_instagram_reactions(
  1230. message, (item.reactions.emojis if item.reactions else [])
  1231. )
  1232. # endregion
  1233. # region Updating portal info
  1234. def _get_thread_name(self, thread: Thread) -> str:
  1235. if self.is_direct:
  1236. if self.other_user_pk == thread.viewer_id and len(thread.users) == 0:
  1237. return "Instagram chat with yourself"
  1238. elif len(thread.users) == 1:
  1239. tpl = self.config["bridge.private_chat_name_template"]
  1240. ui = thread.users[0]
  1241. return tpl.format(displayname=ui.full_name, id=ui.pk, username=ui.username)
  1242. pass
  1243. elif thread.thread_title:
  1244. return self.config["bridge.group_chat_name_template"].format(name=thread.thread_title)
  1245. else:
  1246. return ""
  1247. async def update_info(self, thread: Thread, source: u.User) -> None:
  1248. changed = await self._update_name(self._get_thread_name(thread))
  1249. changed = await self._update_participants(thread.users, source) or changed
  1250. if changed:
  1251. await self.update_bridge_info()
  1252. await self.update()
  1253. # TODO update power levels with thread.admin_user_ids
  1254. async def update_info_from_puppet(self, puppet: p.Puppet | None = None) -> None:
  1255. if not self.is_direct:
  1256. return
  1257. if not puppet:
  1258. puppet = await self.get_dm_puppet()
  1259. await self._update_photo_from_puppet(puppet)
  1260. if self.name and not self.name_set:
  1261. await self._update_name(self.name)
  1262. async def _update_name(self, name: str) -> bool:
  1263. if name and (self.name != name or not self.name_set):
  1264. self.name = name
  1265. if self.mxid:
  1266. try:
  1267. await self.main_intent.set_room_name(self.mxid, name)
  1268. self.name_set = True
  1269. except Exception:
  1270. self.log.exception("Failed to update name")
  1271. self.name_set = False
  1272. return True
  1273. return False
  1274. async def _update_photo_from_puppet(self, puppet: p.Puppet) -> bool:
  1275. if not self.private_chat_portal_meta and not self.encrypted:
  1276. return False
  1277. if self.avatar_set and self.avatar_url == puppet.photo_mxc:
  1278. return False
  1279. self.avatar_url = puppet.photo_mxc
  1280. if self.mxid:
  1281. try:
  1282. await self.main_intent.set_room_avatar(self.mxid, puppet.photo_mxc)
  1283. self.avatar_set = True
  1284. except Exception:
  1285. self.log.exception("Failed to set room avatar")
  1286. self.avatar_set = False
  1287. return True
  1288. async def _update_participants(self, users: list[ThreadUser], source: u.User) -> bool:
  1289. meta_changed = False
  1290. # Make sure puppets who should be here are here
  1291. for user in users:
  1292. puppet = await p.Puppet.get_by_pk(user.pk)
  1293. await puppet.update_info(user, source)
  1294. if self.mxid:
  1295. await puppet.intent_for(self).ensure_joined(self.mxid)
  1296. if puppet.pk == self.other_user_pk:
  1297. meta_changed = await self._update_photo_from_puppet(puppet)
  1298. if self.mxid:
  1299. # Kick puppets who shouldn't be here
  1300. current_members = {int(user.pk) for user in users}
  1301. for user_id in await self.main_intent.get_room_members(self.mxid):
  1302. pk = p.Puppet.get_id_from_mxid(user_id)
  1303. if pk and pk not in current_members and pk != self.other_user_pk:
  1304. await self.main_intent.kick_user(
  1305. self.mxid,
  1306. p.Puppet.get_mxid_from_id(pk),
  1307. reason="User had left this Instagram DM",
  1308. )
  1309. return meta_changed
  1310. async def _update_read_receipts(self, receipts: dict[int | str, ThreadUserLastSeenAt]) -> None:
  1311. for user_id, receipt in receipts.items():
  1312. message: DBMessage | DBReaction
  1313. message = await DBMessage.get_by_item_id(receipt.item_id, self.receiver)
  1314. if not message:
  1315. reaction: DBReaction
  1316. message, reaction = await asyncio.gather(
  1317. DBMessage.get_closest(self.mxid, int(receipt.timestamp)),
  1318. DBReaction.get_closest(self.mxid, receipt.timestamp_ms),
  1319. )
  1320. if (not message or not message.mxid) and not reaction:
  1321. self.log.debug(
  1322. "Couldn't find message %s to mark as read by %s", receipt, user_id
  1323. )
  1324. continue
  1325. elif not message or (reaction and reaction.mx_timestamp > message.ig_timestamp_ms):
  1326. message = reaction
  1327. puppet = await p.Puppet.get_by_pk(int(user_id), create=False)
  1328. if not puppet:
  1329. continue
  1330. try:
  1331. await puppet.intent_for(self).mark_read(message.mx_room, message.mxid)
  1332. except Exception:
  1333. self.log.warning(
  1334. f"Failed to mark {message.mxid} in {message.mx_room} "
  1335. f"as read by {puppet.intent.mxid}",
  1336. exc_info=True,
  1337. )
  1338. async def get_dm_puppet(self) -> p.Puppet | None:
  1339. if not self.is_direct:
  1340. return None
  1341. return await p.Puppet.get_by_pk(self.other_user_pk)
  1342. # endregion
  1343. # region Backfilling
  1344. async def backfill(self, source: u.User, thread: Thread, is_initial: bool = False) -> None:
  1345. limit = (
  1346. self.config["bridge.backfill.initial_limit"]
  1347. if is_initial
  1348. else self.config["bridge.backfill.missed_limit"]
  1349. )
  1350. if limit == 0:
  1351. return
  1352. elif limit < 0:
  1353. limit = None
  1354. with self.backfill_lock:
  1355. await self._backfill(source, thread, is_initial, limit)
  1356. async def _backfill(
  1357. self, source: u.User, thread: Thread, is_initial: bool, limit: int
  1358. ) -> None:
  1359. self.log.debug("Backfilling history through %s", source.mxid)
  1360. entries = await self._fetch_backfill_items(source, thread, is_initial, limit)
  1361. if not entries:
  1362. self.log.debug("Didn't get any items to backfill from server")
  1363. return
  1364. self.log.debug("Got %d entries from server", len(entries))
  1365. self._backfill_leave = set()
  1366. async with NotificationDisabler(self.mxid, source):
  1367. for entry in reversed(entries):
  1368. sender = await p.Puppet.get_by_pk(int(entry.user_id))
  1369. await self.handle_instagram_item(source, sender, entry, is_backfill=True)
  1370. for intent in self._backfill_leave:
  1371. self.log.trace("Leaving room with %s post-backfill", intent.mxid)
  1372. await intent.leave_room(self.mxid)
  1373. self._backfill_leave = None
  1374. self.log.info("Backfilled %d messages through %s", len(entries), source.mxid)
  1375. async def _fetch_backfill_items(
  1376. self, source: u.User, thread: Thread, is_initial: bool, limit: int
  1377. ) -> list[ThreadItem]:
  1378. items = []
  1379. excluded_count = 0
  1380. self.log.debug("Fetching up to %d messages through %s", limit, source.igpk)
  1381. async for item in source.client.iter_thread(self.thread_id, start_at=thread):
  1382. if len(items) - excluded_count >= limit:
  1383. self.log.debug(f"Fetched {len(items)} messages (the limit)")
  1384. break
  1385. elif not is_initial:
  1386. msg = await DBMessage.get_by_item_id(item.item_id, receiver=self.receiver)
  1387. if msg is not None:
  1388. self.log.debug(
  1389. f"Fetched {len(items)} messages and hit a message"
  1390. " that's already in the database."
  1391. )
  1392. break
  1393. elif not item.is_handleable:
  1394. self.log.debug(
  1395. f"Not counting {item.unhandleable_type} item {item.item_id}"
  1396. " against backfill limit"
  1397. )
  1398. excluded_count += 1
  1399. items.append(item)
  1400. return items
  1401. # endregion
  1402. # region Bridge info state event
  1403. @property
  1404. def bridge_info_state_key(self) -> str:
  1405. return f"net.maunium.instagram://instagram/{self.thread_id}"
  1406. @property
  1407. def bridge_info(self) -> dict[str, Any]:
  1408. return {
  1409. "bridgebot": self.az.bot_mxid,
  1410. "creator": self.main_intent.mxid,
  1411. "protocol": {
  1412. "id": "instagram",
  1413. "displayname": "Instagram DM",
  1414. "avatar_url": self.config["appservice.bot_avatar"],
  1415. },
  1416. "channel": {
  1417. "id": self.thread_id,
  1418. "displayname": self.name,
  1419. "avatar_url": self.avatar_url,
  1420. },
  1421. }
  1422. async def update_bridge_info(self) -> None:
  1423. if not self.mxid:
  1424. self.log.debug("Not updating bridge info: no Matrix room created")
  1425. return
  1426. try:
  1427. self.log.debug("Updating bridge info...")
  1428. await self.main_intent.send_state_event(
  1429. self.mxid, StateBridge, self.bridge_info, self.bridge_info_state_key
  1430. )
  1431. # TODO remove this once https://github.com/matrix-org/matrix-doc/pull/2346 is in spec
  1432. await self.main_intent.send_state_event(
  1433. self.mxid, StateHalfShotBridge, self.bridge_info, self.bridge_info_state_key
  1434. )
  1435. except Exception:
  1436. self.log.warning("Failed to update bridge info", exc_info=True)
  1437. # endregion
  1438. # region Creating Matrix rooms
  1439. async def create_matrix_room(self, source: u.User, info: Thread) -> RoomID | None:
  1440. if self.mxid:
  1441. try:
  1442. await self.update_matrix_room(source, info)
  1443. except Exception:
  1444. self.log.exception("Failed to update portal")
  1445. return self.mxid
  1446. async with self._create_room_lock:
  1447. return await self._create_matrix_room(source, info)
  1448. def _get_invite_content(self, double_puppet: p.Puppet | None) -> dict[str, bool]:
  1449. invite_content = {}
  1450. if double_puppet:
  1451. invite_content["fi.mau.will_auto_accept"] = True
  1452. if self.is_direct:
  1453. invite_content["is_direct"] = True
  1454. return invite_content
  1455. async def update_matrix_room(
  1456. self, source: u.User, info: Thread, backfill: bool = False
  1457. ) -> None:
  1458. puppet = await p.Puppet.get_by_custom_mxid(source.mxid)
  1459. await self.main_intent.invite_user(
  1460. self.mxid,
  1461. source.mxid,
  1462. check_cache=True,
  1463. extra_content=self._get_invite_content(puppet),
  1464. )
  1465. if puppet:
  1466. did_join = await puppet.intent.ensure_joined(self.mxid)
  1467. if did_join and self.is_direct:
  1468. await source.update_direct_chats({self.main_intent.mxid: [self.mxid]})
  1469. await self.update_info(info, source)
  1470. if backfill:
  1471. last_msg = await DBMessage.get_by_item_id(
  1472. info.last_permanent_item.item_id, receiver=self.receiver
  1473. )
  1474. if last_msg is None:
  1475. self.log.debug(
  1476. f"Last permanent item ({info.last_permanent_item.item_id})"
  1477. " not found in database, starting backfilling"
  1478. )
  1479. await self.backfill(source, thread=info, is_initial=False)
  1480. await self._update_read_receipts(info.last_seen_at)
  1481. async def _create_matrix_room(self, source: u.User, info: Thread) -> RoomID | None:
  1482. if self.mxid:
  1483. await self.update_matrix_room(source, info)
  1484. return self.mxid
  1485. await self.update_info(info, source)
  1486. self.log.debug("Creating Matrix room")
  1487. name: str | None = None
  1488. initial_state = [
  1489. {
  1490. "type": str(StateBridge),
  1491. "state_key": self.bridge_info_state_key,
  1492. "content": self.bridge_info,
  1493. },
  1494. # TODO remove this once https://github.com/matrix-org/matrix-doc/pull/2346 is in spec
  1495. {
  1496. "type": str(StateHalfShotBridge),
  1497. "state_key": self.bridge_info_state_key,
  1498. "content": self.bridge_info,
  1499. },
  1500. ]
  1501. invites = []
  1502. if self.config["bridge.encryption.default"] and self.matrix.e2ee:
  1503. self.encrypted = True
  1504. initial_state.append(
  1505. {
  1506. "type": "m.room.encryption",
  1507. "content": self.get_encryption_state_event_json(),
  1508. }
  1509. )
  1510. if self.is_direct:
  1511. invites.append(self.az.bot_mxid)
  1512. if self.encrypted or self.private_chat_portal_meta or not self.is_direct:
  1513. name = self.name
  1514. # We lock backfill lock here so any messages that come between the room being created
  1515. # and the initial backfill finishing wouldn't be bridged before the backfill messages.
  1516. with self.backfill_lock:
  1517. creation_content = {}
  1518. if not self.config["bridge.federate_rooms"]:
  1519. creation_content["m.federate"] = False
  1520. self.mxid = await self.main_intent.create_room(
  1521. name=name,
  1522. is_direct=self.is_direct,
  1523. initial_state=initial_state,
  1524. invitees=invites,
  1525. creation_content=creation_content,
  1526. )
  1527. if not self.mxid:
  1528. raise Exception("Failed to create room: no mxid returned")
  1529. if self.encrypted and self.matrix.e2ee and self.is_direct:
  1530. try:
  1531. await self.az.intent.ensure_joined(self.mxid)
  1532. except Exception:
  1533. self.log.warning(f"Failed to add bridge bot to new private chat {self.mxid}")
  1534. await self.update()
  1535. self.log.debug(f"Matrix room created: {self.mxid}")
  1536. self.by_mxid[self.mxid] = self
  1537. puppet = await p.Puppet.get_by_custom_mxid(source.mxid)
  1538. await self.main_intent.invite_user(
  1539. self.mxid, source.mxid, extra_content=self._get_invite_content(puppet)
  1540. )
  1541. if puppet:
  1542. try:
  1543. if self.is_direct:
  1544. await source.update_direct_chats({self.main_intent.mxid: [self.mxid]})
  1545. await puppet.intent.join_room_by_id(self.mxid)
  1546. except MatrixError:
  1547. self.log.debug(
  1548. "Failed to join custom puppet into newly created portal", exc_info=True
  1549. )
  1550. await self._update_participants(info.users, source)
  1551. try:
  1552. await self.backfill(source, thread=info, is_initial=True)
  1553. except Exception:
  1554. self.log.exception("Failed to backfill new portal")
  1555. await self._update_read_receipts(info.last_seen_at)
  1556. return self.mxid
  1557. # endregion
  1558. # region Database getters
  1559. async def postinit(self) -> None:
  1560. self.by_thread_id[(self.thread_id, self.receiver)] = self
  1561. if self.mxid:
  1562. self.by_mxid[self.mxid] = self
  1563. self._main_intent = (
  1564. (await p.Puppet.get_by_pk(self.other_user_pk)).default_mxid_intent
  1565. if self.other_user_pk
  1566. else self.az.intent
  1567. )
  1568. async def delete(self) -> None:
  1569. await DBMessage.delete_all(self.mxid)
  1570. self.by_mxid.pop(self.mxid, None)
  1571. self.mxid = None
  1572. self.encrypted = False
  1573. await self.update()
  1574. async def save(self) -> None:
  1575. await self.update()
  1576. @classmethod
  1577. def all_with_room(cls) -> AsyncGenerator[Portal, None]:
  1578. return cls._db_to_portals(super().all_with_room())
  1579. @classmethod
  1580. def find_private_chats_with(cls, other_user: int) -> AsyncGenerator[Portal, None]:
  1581. return cls._db_to_portals(super().find_private_chats_with(other_user))
  1582. @classmethod
  1583. async def find_private_chat(cls, receiver: int, other_user: int) -> Portal | None:
  1584. thread_id = await super().find_private_chat_id(receiver, other_user)
  1585. if not thread_id:
  1586. return None
  1587. return await cls.get_by_thread_id(thread_id, receiver=receiver, is_group=False)
  1588. @classmethod
  1589. async def _db_to_portals(cls, query: Awaitable[list[Portal]]) -> AsyncGenerator[Portal, None]:
  1590. portals = await query
  1591. for index, portal in enumerate(portals):
  1592. try:
  1593. yield cls.by_thread_id[(portal.thread_id, portal.receiver)]
  1594. except KeyError:
  1595. await portal.postinit()
  1596. yield portal
  1597. @classmethod
  1598. @async_getter_lock
  1599. async def get_by_mxid(cls, mxid: RoomID) -> Portal | None:
  1600. try:
  1601. return cls.by_mxid[mxid]
  1602. except KeyError:
  1603. pass
  1604. portal = cast(cls, await super().get_by_mxid(mxid))
  1605. if portal is not None:
  1606. await portal.postinit()
  1607. return portal
  1608. return None
  1609. @classmethod
  1610. @async_getter_lock
  1611. async def get_by_thread_id(
  1612. cls,
  1613. thread_id: str,
  1614. *,
  1615. receiver: int,
  1616. is_group: bool | None = None,
  1617. other_user_pk: int | None = None,
  1618. ) -> Portal | None:
  1619. if is_group and receiver != 0:
  1620. receiver = 0
  1621. try:
  1622. return cls.by_thread_id[(thread_id, receiver)]
  1623. except KeyError:
  1624. pass
  1625. if is_group is None and receiver != 0:
  1626. try:
  1627. return cls.by_thread_id[(thread_id, 0)]
  1628. except KeyError:
  1629. pass
  1630. portal = cast(
  1631. cls,
  1632. await super().get_by_thread_id(
  1633. thread_id, receiver=receiver, rec_must_match=is_group is not None
  1634. ),
  1635. )
  1636. if portal is not None:
  1637. await portal.postinit()
  1638. return portal
  1639. if is_group is not None:
  1640. portal = cls(thread_id, receiver, other_user_pk=other_user_pk)
  1641. await portal.insert()
  1642. await portal.postinit()
  1643. return portal
  1644. return None
  1645. @classmethod
  1646. async def get_by_thread(cls, thread: Thread, receiver: int) -> Portal | None:
  1647. if thread.is_group:
  1648. receiver = 0
  1649. other_user_pk = None
  1650. else:
  1651. if len(thread.users) == 0:
  1652. other_user_pk = receiver
  1653. else:
  1654. other_user_pk = thread.users[0].pk
  1655. return await cls.get_by_thread_id(
  1656. thread.thread_id,
  1657. receiver=receiver,
  1658. is_group=thread.is_group,
  1659. other_user_pk=other_user_pk,
  1660. )
  1661. # endregion