portal.py 70 KB

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