portal.py 63 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378
  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 typing import (Dict, Tuple, Optional, List, Deque, Set, Any, Union, AsyncGenerator,
  17. Awaitable, Callable, TYPE_CHECKING, cast)
  18. from collections import deque
  19. from io import BytesIO
  20. import mimetypes
  21. import asyncio
  22. import json
  23. import asyncpg
  24. import magic
  25. from mautrix.util.message_send_checkpoint import MessageSendCheckpointStatus
  26. from mauigpapi.types import (Thread, ThreadUser, ThreadItem, RegularMediaItem, MediaType,
  27. ReactionStatus, Reaction, AnimatedMediaItem, ThreadItemType,
  28. VoiceMediaItem, ExpiredMediaItem, MessageSyncMessage, ReelShareType,
  29. TypingStatus, ThreadUserLastSeenAt, MediaShareItem,
  30. ReelMediaShareItem, CommandResponse, ShareVoiceResponse)
  31. from mautrix.appservice import AppService, IntentAPI
  32. from mautrix.bridge import BasePortal, NotificationDisabler, async_getter_lock
  33. from mautrix.types import (EventID, MessageEventContent, RoomID, EventType, MessageType, ImageInfo,
  34. VideoInfo, MediaMessageEventContent, TextMessageEventContent, AudioInfo,
  35. ContentURI, LocationMessageEventContent, Format, UserID)
  36. from mautrix.errors import MatrixError, MForbidden, MNotFound, SessionNotFound
  37. from mautrix.util.simple_lock import SimpleLock
  38. from mautrix.util import ffmpeg
  39. from .db import Portal as DBPortal, Message as DBMessage, Reaction as DBReaction
  40. from .config import Config
  41. from . import user as u, puppet as p, matrix as m
  42. if TYPE_CHECKING:
  43. from .__main__ import InstagramBridge
  44. try:
  45. from mautrix.crypto.attachments import encrypt_attachment, decrypt_attachment
  46. except ImportError:
  47. encrypt_attachment = decrypt_attachment = None
  48. try:
  49. from PIL import Image
  50. except ImportError:
  51. Image = None
  52. StateBridge = EventType.find("m.bridge", EventType.Class.STATE)
  53. StateHalfShotBridge = EventType.find("uk.half-shot.bridge", EventType.Class.STATE)
  54. FileInfo = Union[AudioInfo, ImageInfo, VideoInfo]
  55. MediaData = Union[
  56. AnimatedMediaItem,
  57. ExpiredMediaItem,
  58. MediaShareItem,
  59. ReelMediaShareItem,
  60. RegularMediaItem,
  61. VoiceMediaItem,
  62. ]
  63. MediaUploadFunc = Callable[['u.User', MediaData, IntentAPI], Awaitable[MediaMessageEventContent]]
  64. class Portal(DBPortal, BasePortal):
  65. by_mxid: Dict[RoomID, 'Portal'] = {}
  66. by_thread_id: Dict[Tuple[str, int], 'Portal'] = {}
  67. config: Config
  68. matrix: 'm.MatrixHandler'
  69. az: AppService
  70. private_chat_portal_meta: bool
  71. _main_intent: Optional[IntentAPI]
  72. _create_room_lock: asyncio.Lock
  73. backfill_lock: SimpleLock
  74. _msgid_dedup: Deque[str]
  75. _reqid_dedup: Set[str]
  76. _reaction_dedup: Deque[Tuple[str, int, str]]
  77. _main_intent: IntentAPI
  78. _last_participant_update: Set[int]
  79. _reaction_lock: asyncio.Lock
  80. _backfill_leave: Optional[Set[IntentAPI]]
  81. _typing: Set[UserID]
  82. def __init__(self, thread_id: str, receiver: int, other_user_pk: Optional[int],
  83. mxid: Optional[RoomID] = None, name: Optional[str] = None,
  84. avatar_url: Optional[ContentURI] = None, encrypted: bool = False,
  85. name_set: bool = False, avatar_set: bool = False,
  86. relay_user_id: Optional[UserID] = None) -> None:
  87. super().__init__(thread_id, receiver, other_user_pk, mxid, name, avatar_url, encrypted,
  88. name_set, avatar_set, relay_user_id)
  89. self._create_room_lock = asyncio.Lock()
  90. self.log = self.log.getChild(thread_id)
  91. self._msgid_dedup = deque(maxlen=100)
  92. self._reaction_dedup = deque(maxlen=100)
  93. self._reqid_dedup = set()
  94. self._last_participant_update = set()
  95. self.backfill_lock = SimpleLock("Waiting for backfilling to finish before handling %s",
  96. log=self.log)
  97. self._backfill_leave = None
  98. self._main_intent = None
  99. self._reaction_lock = asyncio.Lock()
  100. self._typing = set()
  101. self._relay_user = None
  102. @property
  103. def is_direct(self) -> bool:
  104. return self.other_user_pk is not None
  105. @property
  106. def main_intent(self) -> IntentAPI:
  107. if not self._main_intent:
  108. raise ValueError("Portal must be postinit()ed before main_intent can be used")
  109. return self._main_intent
  110. @classmethod
  111. def init_cls(cls, bridge: 'InstagramBridge') -> None:
  112. cls.config = bridge.config
  113. cls.matrix = bridge.matrix
  114. cls.az = bridge.az
  115. cls.loop = bridge.loop
  116. cls.bridge = bridge
  117. cls.private_chat_portal_meta = cls.config["bridge.private_chat_portal_meta"]
  118. NotificationDisabler.puppet_cls = p.Puppet
  119. NotificationDisabler.config_enabled = cls.config["bridge.backfill.disable_notifications"]
  120. # region Misc
  121. async def _send_delivery_receipt(self, event_id: EventID) -> None:
  122. if event_id and self.config["bridge.delivery_receipts"]:
  123. try:
  124. await self.az.intent.mark_read(self.mxid, event_id)
  125. except Exception:
  126. self.log.exception("Failed to send delivery receipt for %s", event_id)
  127. async def _send_bridge_error(self, sender: 'u.User', err: Union[Exception, str],
  128. event_id: EventID, event_type: EventType,
  129. message_type: Optional[MessageType] = None,
  130. msg: Optional[str] = None, confirmed: bool = False,
  131. status: MessageSendCheckpointStatus = MessageSendCheckpointStatus.PERM_FAILURE,
  132. ) -> None:
  133. sender.send_remote_checkpoint(
  134. status,
  135. event_id,
  136. self.mxid,
  137. event_type,
  138. message_type=message_type,
  139. error=err,
  140. )
  141. if self.config["bridge.delivery_error_reports"]:
  142. event_type_str = {
  143. EventType.REACTION: "reaction",
  144. EventType.ROOM_REDACTION: "redaction",
  145. }.get(event_type, "message")
  146. error_type = "was not" if confirmed else "may not have been"
  147. await self._send_message(self.main_intent, TextMessageEventContent(
  148. msgtype=MessageType.NOTICE,
  149. body=f"\u26a0 Your {event_type_str} {error_type} bridged: {msg or str(err)}"))
  150. async def _upsert_reaction(self, existing: Optional[DBReaction], intent: IntentAPI,
  151. mxid: EventID, message: DBMessage,
  152. sender: Union['u.User', 'p.Puppet'], reaction: str) -> None:
  153. if existing:
  154. self.log.debug(f"_upsert_reaction redacting {existing.mxid} and inserting {mxid}"
  155. f" (message: {message.mxid})")
  156. await intent.redact(existing.mx_room, existing.mxid)
  157. await existing.edit(reaction=reaction, mxid=mxid, mx_room=message.mx_room)
  158. else:
  159. self.log.debug(f"_upsert_reaction inserting {mxid} (message: {message.mxid})")
  160. await DBReaction(mxid=mxid, mx_room=message.mx_room, ig_item_id=message.item_id,
  161. ig_receiver=self.receiver, ig_sender=sender.igpk, reaction=reaction
  162. ).insert()
  163. # endregion
  164. # region Matrix event handling
  165. def _status_from_exception(self, e: Exception) -> MessageSendCheckpointStatus:
  166. if isinstance(e, NotImplementedError):
  167. return MessageSendCheckpointStatus.UNSUPPORTED
  168. return MessageSendCheckpointStatus.PERM_FAILURE
  169. async def handle_matrix_message(self, sender: 'u.User', message: MessageEventContent,
  170. event_id: EventID) -> None:
  171. try:
  172. await self._handle_matrix_message(sender, message, event_id)
  173. except Exception as e:
  174. self.log.exception(f"Fatal error handling Matrix event {event_id}: {e}")
  175. await self._send_bridge_error(
  176. sender,
  177. e,
  178. event_id,
  179. EventType.ROOM_MESSAGE,
  180. message_type=message.msgtype,
  181. status=self._status_from_exception(e),
  182. confirmed=True,
  183. )
  184. async def _handle_matrix_image(
  185. self, sender: 'u.User', event_id: EventID, request_id: str, data: bytes, mime_type: str,
  186. width: Optional[int] = None, height: Optional[int] = None
  187. ) -> CommandResponse:
  188. if mime_type not in ("image/jpeg", "image/webp"):
  189. with BytesIO(data) as inp, BytesIO() as out:
  190. img = Image.open(inp)
  191. img.convert("RGBA").save(out, format="WEBP")
  192. data = out.getvalue()
  193. mime_type = "image/webp"
  194. self.log.trace(f"Uploading photo from {event_id} (mime: {mime_type})")
  195. upload_resp = await sender.client.upload_photo(
  196. data, mime=mime_type, width=width, height=height
  197. )
  198. self.log.trace(f"Broadcasting uploaded photo with request ID {request_id}")
  199. return await sender.client.broadcast(
  200. self.thread_id,
  201. ThreadItemType.CONFIGURE_PHOTO,
  202. client_context=request_id,
  203. upload_id=upload_resp.upload_id,
  204. allow_full_aspect_ratio="1",
  205. )
  206. async def _handle_matrix_video(
  207. self, sender: 'u.User', event_id: EventID, request_id: str, data: bytes, mime_type: str,
  208. duration: Optional[int] = None, width: Optional[int] = None, height: Optional[int] = None,
  209. ) -> CommandResponse:
  210. if mime_type != "video/mp4":
  211. data = await ffmpeg.convert_bytes(
  212. data,
  213. output_extension=".mp4",
  214. output_args=("-c:v", "libx264", "-c:a", "aac"),
  215. input_mime=mime_type
  216. )
  217. self.log.trace(f"Uploading video from {event_id}")
  218. _, upload_id = await sender.client.upload_mp4(
  219. data, duration_ms=duration, width=width, height=height
  220. )
  221. self.log.trace(f"Broadcasting uploaded video with request ID {request_id}")
  222. return await sender.client.broadcast(
  223. self.thread_id,
  224. ThreadItemType.CONFIGURE_VIDEO,
  225. client_context=request_id,
  226. upload_id=upload_id,
  227. video_result="",
  228. )
  229. async def _handle_matrix_audio(
  230. self, sender: 'u.User', event_id: EventID, request_id: str, data: bytes, mime_type: str,
  231. waveform: List[int], duration: Optional[int] = None,
  232. ) -> CommandResponse:
  233. if mime_type != "audio/mp4":
  234. data = await ffmpeg.convert_bytes(
  235. data,
  236. output_extension=".m4a",
  237. output_args=("-c:a", "aac"),
  238. input_mime=mime_type
  239. )
  240. self.log.trace(f"Uploading audio from {event_id}")
  241. _, upload_id = await sender.client.upload_mp4(data, audio=True, duration_ms=duration)
  242. self.log.trace(f"Broadcasting uploaded audio with request ID {request_id}")
  243. return await sender.client.broadcast_audio(
  244. self.thread_id,
  245. is_direct=self.is_direct,
  246. client_context=request_id,
  247. upload_id=upload_id,
  248. waveform=json.dumps([(part or 0) / 1024 for part in waveform],
  249. separators=(",", ":")),
  250. waveform_sampling_frequency_hz="10",
  251. )
  252. async def _handle_matrix_message(self, orig_sender: 'u.User', message: MessageEventContent,
  253. event_id: EventID) -> None:
  254. sender, is_relay = await self.get_relay_sender(orig_sender, f"message {event_id}")
  255. assert sender, "user is not logged in"
  256. assert sender.is_connected, "You're not connected to Instagram"
  257. if is_relay:
  258. await self.apply_relay_message_format(orig_sender, message)
  259. request_id = sender.state.gen_client_context()
  260. self._reqid_dedup.add(request_id)
  261. self.log.debug(f"Handling Matrix message {event_id} from {sender.mxid}/{sender.igpk} "
  262. f"with request ID {request_id}")
  263. if message.msgtype in (MessageType.EMOTE, MessageType.TEXT):
  264. text = message.body
  265. if message.msgtype == MessageType.EMOTE:
  266. text = f"/me {text}"
  267. self.log.trace(f"Sending Matrix text from {event_id} with request ID {request_id}")
  268. resp = await sender.mqtt.send_text(self.thread_id, text=text,
  269. client_context=request_id)
  270. elif message.msgtype.is_media:
  271. if message.file and decrypt_attachment:
  272. data = await self.main_intent.download_media(message.file.url)
  273. data = decrypt_attachment(data, message.file.key.key,
  274. message.file.hashes.get("sha256"), message.file.iv)
  275. else:
  276. data = await self.main_intent.download_media(message.url)
  277. mime_type = message.info.mimetype or magic.from_buffer(data, mime=True)
  278. if message.msgtype == MessageType.IMAGE:
  279. resp = await self._handle_matrix_image(
  280. sender, event_id, request_id, data, mime_type,
  281. width=message.info.width, height=message.info.height,
  282. )
  283. elif message.msgtype == MessageType.AUDIO:
  284. waveform = message.get("org.matrix.msc1767.audio", {}).get("waveform", [0] * 30)
  285. resp = await self._handle_matrix_audio(
  286. sender, event_id, request_id, data, mime_type,
  287. waveform, duration=message.info.duration,
  288. )
  289. elif message.msgtype == MessageType.VIDEO:
  290. resp = await self._handle_matrix_video(
  291. sender, event_id, request_id, data, mime_type, duration=message.info.duration,
  292. width=message.info.width, height=message.info.height,
  293. )
  294. else:
  295. raise NotImplementedError(
  296. "Non-image/video/audio files are currently not supported"
  297. )
  298. else:
  299. raise NotImplementedError(f"Unknown message type {message.msgtype}")
  300. self.log.trace(f"Got response to message send {request_id}: {resp}")
  301. if resp.status != "ok":
  302. self.log.warning(f"Failed to handle {event_id}: {resp}")
  303. raise Exception(f"Failed to handle event. Error: {resp.payload.message}")
  304. else:
  305. sender.send_remote_checkpoint(
  306. status=MessageSendCheckpointStatus.SUCCESS,
  307. event_id=event_id,
  308. room_id=self.mxid,
  309. event_type=EventType.ROOM_MESSAGE,
  310. message_type=message.msgtype,
  311. )
  312. await self._send_delivery_receipt(event_id)
  313. self._msgid_dedup.appendleft(resp.payload.item_id)
  314. try:
  315. await DBMessage(mxid=event_id, mx_room=self.mxid, item_id=resp.payload.item_id,
  316. receiver=self.receiver, sender=sender.igpk).insert()
  317. except asyncpg.UniqueViolationError as e:
  318. self.log.warning(f"Error while persisting {event_id} "
  319. f"({resp.payload.client_context}) -> {resp.payload.item_id}: {e}")
  320. self._reqid_dedup.remove(request_id)
  321. self.log.debug(f"Handled Matrix message {event_id} ({resp.payload.client_context}) "
  322. f"-> {resp.payload.item_id}")
  323. async def handle_matrix_reaction(self, sender: 'u.User', event_id: EventID,
  324. reacting_to: EventID, emoji: str) -> None:
  325. try:
  326. await self._handle_matrix_reaction(sender, event_id, reacting_to, emoji)
  327. except Exception as e:
  328. self.log.exception(f"Fatal error handling Matrix event {event_id}: {e}")
  329. await self._send_bridge_error(
  330. sender,
  331. e,
  332. event_id,
  333. EventType.REACTION,
  334. status=self._status_from_exception(e),
  335. confirmed=True,
  336. msg="Fatal error handling reaction (see logs for more details)",
  337. )
  338. async def _handle_matrix_reaction(self, sender: 'u.User', event_id: EventID,
  339. reacting_to: EventID, emoji: str) -> None:
  340. message = await DBMessage.get_by_mxid(reacting_to, self.mxid)
  341. if not message or message.is_internal:
  342. self.log.debug(f"Ignoring reaction to unknown event {reacting_to}")
  343. return
  344. if not await sender.is_logged_in():
  345. self.log.debug(f"Ignoring reaction by non-logged-in user {sender.mxid}")
  346. return
  347. existing = await DBReaction.get_by_item_id(message.item_id, message.receiver, sender.igpk)
  348. if existing and existing.reaction == emoji:
  349. sender.send_remote_checkpoint(
  350. status=MessageSendCheckpointStatus.SUCCESS,
  351. event_id=event_id,
  352. room_id=self.mxid,
  353. event_type=EventType.REACTION,
  354. )
  355. return
  356. dedup_id = (message.item_id, sender.igpk, emoji)
  357. self._reaction_dedup.appendleft(dedup_id)
  358. async with self._reaction_lock:
  359. try:
  360. resp = await sender.mqtt.send_reaction(self.thread_id, item_id=message.item_id,
  361. emoji=emoji)
  362. if resp.status != "ok":
  363. raise Exception(f"Failed to react to {event_id}: {resp}")
  364. except Exception as e:
  365. self.log.exception(f"Failed to handle {event_id}: {e}")
  366. raise
  367. else:
  368. sender.send_remote_checkpoint(
  369. status=MessageSendCheckpointStatus.SUCCESS,
  370. event_id=event_id,
  371. room_id=self.mxid,
  372. event_type=EventType.REACTION,
  373. )
  374. await self._send_delivery_receipt(event_id)
  375. self.log.trace(f"{sender.mxid} reacted to {message.item_id} with {emoji}")
  376. await self._upsert_reaction(existing, self.main_intent, event_id, message, sender,
  377. emoji)
  378. async def handle_matrix_redaction(self, orig_sender: 'u.User', event_id: EventID,
  379. redaction_event_id: EventID) -> None:
  380. sender = None
  381. try:
  382. sender, _ = await self.get_relay_sender(orig_sender, f"redaction {event_id}")
  383. if not sender:
  384. raise Exception("User is not logged in")
  385. await self._handle_matrix_redaction(sender, event_id, redaction_event_id)
  386. except Exception as e:
  387. self.log.exception(f"Fatal error handling Matrix event {event_id}: {e}")
  388. await self._send_bridge_error(
  389. sender or orig_sender,
  390. e,
  391. redaction_event_id,
  392. EventType.ROOM_REDACTION,
  393. status=self._status_from_exception(e),
  394. confirmed=True,
  395. )
  396. async def _handle_matrix_redaction(self, sender: 'u.User', event_id: EventID,
  397. redaction_event_id: EventID) -> None:
  398. if not sender.is_connected:
  399. raise Exception("You're not connected to Instagram")
  400. reaction = await DBReaction.get_by_mxid(event_id, self.mxid)
  401. if reaction:
  402. try:
  403. await reaction.delete()
  404. await sender.mqtt.send_reaction(self.thread_id, item_id=reaction.ig_item_id,
  405. reaction_status=ReactionStatus.DELETED, emoji="")
  406. except Exception as e:
  407. raise Exception(f"Removing reaction failed: {e}")
  408. else:
  409. sender.send_remote_checkpoint(
  410. status=MessageSendCheckpointStatus.SUCCESS,
  411. event_id=redaction_event_id,
  412. room_id=self.mxid,
  413. event_type=EventType.ROOM_REDACTION,
  414. )
  415. await self._send_delivery_receipt(redaction_event_id)
  416. self.log.trace(f"Removed {reaction} after Matrix redaction")
  417. return
  418. message = await DBMessage.get_by_mxid(event_id, self.mxid)
  419. if message and not message.is_internal:
  420. try:
  421. await message.delete()
  422. await sender.client.delete_item(self.thread_id, message.item_id)
  423. self.log.trace(f"Removed {message} after Matrix redaction")
  424. except Exception as e:
  425. raise Exception(f"Removing message failed: {e}")
  426. else:
  427. sender.send_remote_checkpoint(
  428. status=MessageSendCheckpointStatus.SUCCESS,
  429. event_id=redaction_event_id,
  430. room_id=self.mxid,
  431. event_type=EventType.ROOM_REDACTION,
  432. )
  433. await self._send_delivery_receipt(redaction_event_id)
  434. self.log.trace(f"Removed {reaction} after Matrix redaction")
  435. return
  436. raise Exception("No message or reaction found for redaction")
  437. async def handle_matrix_typing(self, users: Set[UserID]) -> None:
  438. if users == self._typing:
  439. return
  440. old_typing = self._typing
  441. self._typing = users
  442. await self._handle_matrix_typing(old_typing - users, TypingStatus.OFF)
  443. await self._handle_matrix_typing(users - old_typing, TypingStatus.TEXT)
  444. async def _handle_matrix_typing(self, users: Set[UserID], status: TypingStatus) -> None:
  445. for mxid in users:
  446. user = await u.User.get_by_mxid(mxid, create=False)
  447. if ((not user or not await user.is_logged_in() or user.remote_typing_status == status
  448. or not user.is_connected)):
  449. continue
  450. user.remote_typing_status = None
  451. await user.mqtt.indicate_activity(self.thread_id, status)
  452. async def handle_matrix_leave(self, user: 'u.User') -> None:
  453. if not await user.is_logged_in():
  454. return
  455. if self.is_direct:
  456. self.log.info(f"{user.mxid} left private chat portal with {self.other_user_pk}")
  457. if user.igpk == self.receiver:
  458. self.log.info(f"{user.mxid} was the recipient of this portal. "
  459. "Cleaning up and deleting...")
  460. await self.cleanup_and_delete()
  461. else:
  462. self.log.debug(f"{user.mxid} left portal to {self.thread_id}")
  463. # TODO cleanup if empty
  464. # endregion
  465. # region Instagram event handling
  466. async def _reupload_instagram_media(self, source: 'u.User', media: RegularMediaItem,
  467. intent: IntentAPI) -> MediaMessageEventContent:
  468. if media.media_type == MediaType.IMAGE:
  469. image = media.best_image
  470. if not image:
  471. raise ValueError("Attachment not available: didn't find photo URL")
  472. url = image.url
  473. msgtype = MessageType.IMAGE
  474. info = ImageInfo(height=image.height, width=image.width)
  475. elif media.media_type == MediaType.VIDEO:
  476. video = media.best_video
  477. if not video:
  478. raise ValueError("Attachment not available: didn't find video URL")
  479. url = video.url
  480. msgtype = MessageType.VIDEO
  481. info = VideoInfo(height=video.height, width=video.width)
  482. else:
  483. raise ValueError("Attachment not available: unsupported media type")
  484. return await self._reupload_instagram_file(source, url, msgtype, info, intent)
  485. async def _reupload_instagram_animated(self, source: 'u.User', media: AnimatedMediaItem,
  486. intent: IntentAPI) -> MediaMessageEventContent:
  487. url = media.images.fixed_height.webp
  488. info = ImageInfo(height=int(media.images.fixed_height.height),
  489. width=int(media.images.fixed_height.width))
  490. return await self._reupload_instagram_file(source, url, MessageType.IMAGE, info, intent)
  491. async def _reupload_instagram_voice(self, source: 'u.User', media: VoiceMediaItem,
  492. intent: IntentAPI) -> MediaMessageEventContent:
  493. async def convert_to_ogg(data, mimetype):
  494. converted = await ffmpeg.convert_bytes(data, ".ogg", output_args=('-c:a', 'libvorbis'),
  495. input_mime=mimetype)
  496. return converted, "audio/ogg"
  497. url = media.media.audio.audio_src
  498. info = AudioInfo(duration=media.media.audio.duration)
  499. waveform = [int(p * 1024) for p in media.media.audio.waveform_data]
  500. content = await self._reupload_instagram_file(
  501. source, url, MessageType.AUDIO, info, intent, convert_to_ogg
  502. )
  503. content["org.matrix.msc1767.file"] = {
  504. "url": content.url,
  505. "name": content.body,
  506. **(content.file.serialize() if content.file else {}),
  507. **(content.info.serialize() if content.info else {}),
  508. }
  509. content["org.matrix.msc1767.audio"] = {
  510. "duration": media.media.audio.duration,
  511. "waveform": waveform,
  512. }
  513. content["org.matrix.msc3245.voice"] = {}
  514. return content
  515. async def _reupload_instagram_file(
  516. self, source: 'u.User', url: str, msgtype: MessageType, info: FileInfo, intent: IntentAPI,
  517. convert_fn: Optional[Callable[[bytes, str], Awaitable[Tuple[bytes, str]]]] = None,
  518. ) -> MediaMessageEventContent:
  519. async with await source.client.raw_http_get(url) as resp:
  520. try:
  521. length = int(resp.headers["Content-Length"])
  522. except KeyError:
  523. # TODO can the download be short-circuited if there's too much data?
  524. self.log.warning("Got file download response with no Content-Length header,"
  525. "reading data dangerously")
  526. length = 0
  527. if length > self.matrix.media_config.upload_size:
  528. self.log.debug(f"{url} was too large ({length} "
  529. f"> {self.matrix.media_config.upload_size})")
  530. raise ValueError("Attachment not available: too large")
  531. data = await resp.read()
  532. info.mimetype = resp.headers["Content-Type"] or magic.from_buffer(data, mime=True)
  533. # Run the conversion function on the data.
  534. if convert_fn is not None:
  535. data, info.mimetype = await convert_fn(data, info.mimetype)
  536. info.size = len(data)
  537. extension = {
  538. "image/webp": ".webp",
  539. "image/jpeg": ".jpg",
  540. "video/mp4": ".mp4",
  541. "audio/mp4": ".m4a",
  542. "audio/ogg": ".ogg",
  543. }.get(info.mimetype)
  544. extension = extension or mimetypes.guess_extension(info.mimetype) or ""
  545. file_name = f"{msgtype.value[2:]}{extension}"
  546. upload_mime_type = info.mimetype
  547. upload_file_name = file_name
  548. decryption_info = None
  549. if self.encrypted and encrypt_attachment:
  550. data, decryption_info = encrypt_attachment(data)
  551. upload_mime_type = "application/octet-stream"
  552. upload_file_name = None
  553. mxc = await intent.upload_media(data, mime_type=upload_mime_type,
  554. filename=upload_file_name)
  555. if decryption_info:
  556. decryption_info.url = mxc
  557. mxc = None
  558. return MediaMessageEventContent(
  559. body=file_name,
  560. external_url=url,
  561. url=mxc,
  562. file=decryption_info,
  563. info=info,
  564. msgtype=msgtype,
  565. )
  566. def _get_instagram_media_info(self, item: ThreadItem) -> Tuple[MediaUploadFunc, MediaData]:
  567. # TODO maybe use a dict and item.item_type instead of a ton of ifs
  568. method = self._reupload_instagram_media
  569. if item.media:
  570. media_data = item.media
  571. elif item.visual_media:
  572. media_data = item.visual_media.media
  573. elif item.animated_media:
  574. media_data = item.animated_media
  575. method = self._reupload_instagram_animated
  576. elif item.voice_media:
  577. media_data = item.voice_media
  578. method = self._reupload_instagram_voice
  579. elif item.reel_share:
  580. media_data = item.reel_share.media
  581. elif item.story_share:
  582. media_data = item.story_share.media
  583. elif item.clip:
  584. media_data = item.clip.clip
  585. elif item.felix_share and item.felix_share.video:
  586. media_data = item.felix_share.video
  587. elif item.media_share:
  588. media_data = item.media_share
  589. else:
  590. self.log.debug(f"Unknown media type in {item}")
  591. raise ValueError("Attachment not available: unsupported media type")
  592. if not media_data:
  593. self.log.debug(f"Didn't get media_data in {item}")
  594. raise ValueError("Attachment not available: unsupported media type")
  595. elif isinstance(media_data, ExpiredMediaItem):
  596. self.log.debug(f"Expired media in item {item}")
  597. raise ValueError("Attachment not available: media expired")
  598. return method, media_data
  599. async def _handle_instagram_media(self, source: 'u.User', intent: IntentAPI, item: ThreadItem
  600. ) -> Optional[EventID]:
  601. try:
  602. reupload_func, media_data = self._get_instagram_media_info(item)
  603. content = await reupload_func(source, media_data, intent)
  604. except ValueError as e:
  605. content = TextMessageEventContent(body=str(e), msgtype=MessageType.NOTICE)
  606. except Exception:
  607. self.log.warning("Failed to upload media", exc_info=True)
  608. content = TextMessageEventContent(body="Attachment not available: failed to copy file",
  609. msgtype=MessageType.NOTICE)
  610. await self._add_instagram_reply(content, item.replied_to_message)
  611. return await self._send_message(intent, content, timestamp=item.timestamp // 1000)
  612. async def _handle_instagram_media_share(self, source: 'u.User', intent: IntentAPI,
  613. item: ThreadItem) -> Optional[EventID]:
  614. if item.media_share:
  615. share_item = item.media_share
  616. item_type_name = share_item.media_type.human_name
  617. elif item.clip:
  618. share_item = item.clip.clip
  619. item_type_name = "clip"
  620. elif item.felix_share and item.felix_share.video:
  621. share_item = item.felix_share.video
  622. item_type_name = share_item.media_type.human_name
  623. elif item.story_share:
  624. share_item = item.story_share.media
  625. item_type_name = "story"
  626. else:
  627. return None
  628. user_text = f"@{share_item.user.username}"
  629. user_link = (f'<a href="https://www.instagram.com/{share_item.user.username}/">'
  630. f'{user_text}</a>')
  631. prefix = TextMessageEventContent(msgtype=MessageType.NOTICE, format=Format.HTML,
  632. body=f"Sent {user_text}'s {item_type_name}",
  633. formatted_body=f"Sent {user_link}'s {item_type_name}")
  634. await self._send_message(intent, prefix, timestamp=item.timestamp // 1000)
  635. event_id = await self._handle_instagram_media(source, intent, item)
  636. if share_item.caption:
  637. external_url = f"https://www.instagram.com/p/{share_item.code}/"
  638. body = (f"> {share_item.caption.user.username}: {share_item.caption.text}\n\n"
  639. f"{external_url}")
  640. formatted_body = (f"<blockquote><strong>{share_item.caption.user.username}</strong>"
  641. f" {share_item.caption.text}</blockquote>"
  642. f'<a href="{external_url}">instagram.com/p/{share_item.code}</a>')
  643. caption = TextMessageEventContent(msgtype=MessageType.TEXT, body=body,
  644. formatted_body=formatted_body, format=Format.HTML,
  645. external_url=external_url)
  646. await self._send_message(intent, caption, timestamp=item.timestamp // 1000)
  647. return event_id
  648. async def _handle_instagram_reel_share(self, source: 'u.User', intent: IntentAPI,
  649. item: ThreadItem) -> Optional[EventID]:
  650. media = item.reel_share.media
  651. prefix_html = None
  652. if item.reel_share.type == ReelShareType.REPLY:
  653. if item.reel_share.reel_owner_id == source.igpk:
  654. prefix = "Replied to your story"
  655. else:
  656. username = media.user.username
  657. prefix = f"Sent @{username}'s story"
  658. user_link = f'<a href="https://www.instagram.com/{username}/">@{username}</a>'
  659. prefix_html = f"Sent {user_link}'s story"
  660. elif item.reel_share.type == ReelShareType.REACTION:
  661. prefix = "Reacted to your story"
  662. elif item.reel_share.type == ReelShareType.MENTION:
  663. if item.reel_share.mentioned_user_id == source.igpk:
  664. prefix = "Mentioned you in their story"
  665. else:
  666. prefix = "You mentioned them in your story"
  667. else:
  668. self.log.debug(f"Unsupported reel share type {item.reel_share.type}")
  669. return None
  670. prefix_content = TextMessageEventContent(msgtype=MessageType.NOTICE, body=prefix)
  671. if prefix_html:
  672. prefix_content.format = Format.HTML
  673. prefix_content.formatted_body = prefix_html
  674. content = TextMessageEventContent(msgtype=MessageType.TEXT, body=item.reel_share.text)
  675. if not content.body and isinstance(media, MediaShareItem):
  676. content.body = media.caption.text if media.caption else ""
  677. if not content.body:
  678. content.body = "<no caption>"
  679. await self._send_message(intent, prefix_content, timestamp=item.timestamp // 1000)
  680. if isinstance(media, ExpiredMediaItem):
  681. # TODO send message about expired story
  682. pass
  683. else:
  684. fake_item_id = f"fi.mau.instagram.reel_share.{item.user_id}.{media.pk}"
  685. existing = await DBMessage.get_by_item_id(fake_item_id, self.receiver)
  686. if existing:
  687. # If the user already reacted or replied to the same reel share item,
  688. # use a Matrix reply instead of reposting the image.
  689. content.set_reply(existing.mxid)
  690. else:
  691. media_event_id = await self._handle_instagram_media(source, intent, item)
  692. await DBMessage(mxid=media_event_id, mx_room=self.mxid, item_id=fake_item_id,
  693. receiver=self.receiver, sender=media.user.pk).insert()
  694. return await self._send_message(intent, content, timestamp=item.timestamp // 1000)
  695. async def _handle_instagram_text(self, intent: IntentAPI, item: ThreadItem, text: str,
  696. ) -> EventID:
  697. content = TextMessageEventContent(msgtype=MessageType.TEXT, body=text)
  698. await self._add_instagram_reply(content, item.replied_to_message)
  699. return await self._send_message(intent, content, timestamp=item.timestamp // 1000)
  700. async def _send_instagram_unhandled(self, intent: IntentAPI, item: ThreadItem) -> EventID:
  701. content = TextMessageEventContent(msgtype=MessageType.NOTICE,
  702. body=f"Unsupported message type {item.item_type.value}")
  703. await self._add_instagram_reply(content, item.replied_to_message)
  704. return await self._send_message(intent, content, timestamp=item.timestamp // 1000)
  705. async def _handle_instagram_location(self, intent: IntentAPI, item: ThreadItem
  706. ) -> Optional[EventID]:
  707. loc = item.location
  708. if not loc.lng or not loc.lat:
  709. # TODO handle somehow
  710. return None
  711. long_char = "E" if loc.lng > 0 else "W"
  712. lat_char = "N" if loc.lat > 0 else "S"
  713. body = (f"{loc.name} - {round(abs(loc.lat), 4)}° {lat_char}, "
  714. f"{round(abs(loc.lng), 4)}° {long_char}")
  715. url = f"https://www.openstreetmap.org/#map=15/{loc.lat}/{loc.lng}"
  716. external_url = None
  717. if loc.external_source == "facebook_places":
  718. external_url = f"https://www.facebook.com/{loc.short_name}-{loc.facebook_places_id}"
  719. content = LocationMessageEventContent(
  720. msgtype=MessageType.LOCATION, geo_uri=f"geo:{loc.lat},{loc.lng}",
  721. body=f"Location: {body}\n{url}", external_url=external_url)
  722. content["format"] = str(Format.HTML)
  723. content["formatted_body"] = f"Location: <a href='{url}'>{body}</a>"
  724. await self._add_instagram_reply(content, item.replied_to_message)
  725. return await self._send_message(intent, content, timestamp=item.timestamp // 1000)
  726. async def _handle_instagram_profile(self, intent: IntentAPI, item: ThreadItem
  727. ) -> Optional[EventID]:
  728. username = item.profile.username
  729. user_link = f'<a href="https://www.instagram.com/{username}/">@{username}</a>'
  730. text = f"Shared @{username}'s profile"
  731. html = f"Shared {user_link}'s profile"
  732. content = TextMessageEventContent(msgtype=MessageType.TEXT, format=Format.HTML,
  733. body=text, formatted_body=html)
  734. await self._add_instagram_reply(content, item.replied_to_message)
  735. return await self._send_message(intent, content, timestamp=item.timestamp // 1000)
  736. async def handle_instagram_item(self, source: 'u.User', sender: 'p.Puppet', item: ThreadItem,
  737. is_backfill: bool = False) -> None:
  738. try:
  739. await self._handle_instagram_item(source, sender, item, is_backfill)
  740. except Exception:
  741. self.log.exception("Fatal error handling Instagram item")
  742. self.log.trace("Item content: %s", item.serialize())
  743. async def _add_instagram_reply(self, content: MessageEventContent,
  744. reply_to: Optional[ThreadItem]) -> None:
  745. if not reply_to:
  746. return
  747. message = await DBMessage.get_by_item_id(reply_to.item_id, self.receiver)
  748. if not message:
  749. return
  750. content.set_reply(message.mxid)
  751. if not isinstance(content, TextMessageEventContent):
  752. return
  753. try:
  754. evt = await self.main_intent.get_event(message.mx_room, message.mxid)
  755. except (MNotFound, MForbidden):
  756. evt = None
  757. if not evt:
  758. return
  759. if evt.type == EventType.ROOM_ENCRYPTED:
  760. try:
  761. evt = await self.matrix.e2ee.decrypt(evt, wait_session_timeout=0)
  762. except SessionNotFound:
  763. return
  764. if isinstance(evt.content, TextMessageEventContent):
  765. evt.content.trim_reply_fallback()
  766. content.set_reply(evt)
  767. async def _handle_instagram_item(self, source: 'u.User', sender: 'p.Puppet', item: ThreadItem,
  768. is_backfill: bool = False) -> None:
  769. if not isinstance(item, ThreadItem):
  770. # Parsing these items failed, they should have been logged already
  771. return
  772. elif item.client_context in self._reqid_dedup:
  773. self.log.debug(f"Ignoring message {item.item_id} ({item.client_context}) by "
  774. f"{item.user_id} as it was sent by us (client_context in dedup queue)")
  775. elif item.item_id in self._msgid_dedup:
  776. self.log.debug(f"Ignoring message {item.item_id} ({item.client_context}) by "
  777. f"{item.user_id} as it was already handled (message.id in dedup queue)")
  778. elif await DBMessage.get_by_item_id(item.item_id, self.receiver) is not None:
  779. self.log.debug(f"Ignoring message {item.item_id} ({item.client_context}) by "
  780. f"{item.user_id} as it was already handled (message.id in database)")
  781. else:
  782. self.log.debug(f"Starting handling of message {item.item_id} ({item.client_context}) "
  783. f"by {item.user_id}")
  784. self._msgid_dedup.appendleft(item.item_id)
  785. if self.backfill_lock.locked and sender.need_backfill_invite(self):
  786. self.log.debug("Adding %s's default puppet to room for backfilling", sender.mxid)
  787. if self.is_direct:
  788. await self.main_intent.invite_user(self.mxid, sender.default_mxid)
  789. intent = sender.default_mxid_intent
  790. await intent.ensure_joined(self.mxid)
  791. self._backfill_leave.add(intent)
  792. else:
  793. intent = sender.intent_for(self)
  794. event_id = None
  795. if item.media or item.animated_media or item.voice_media or item.visual_media:
  796. event_id = await self._handle_instagram_media(source, intent, item)
  797. elif item.location:
  798. event_id = await self._handle_instagram_location(intent, item)
  799. elif item.profile:
  800. event_id = await self._handle_instagram_profile(intent, item)
  801. elif item.reel_share:
  802. event_id = await self._handle_instagram_reel_share(source, intent, item)
  803. elif item.media_share or item.story_share or item.clip or item.felix_share:
  804. event_id = await self._handle_instagram_media_share(source, intent, item)
  805. elif item.action_log:
  806. # These probably don't need to be bridged
  807. self.log.debug(f"Ignoring action log message {item.item_id}")
  808. return
  809. # TODO handle item.clip?
  810. if item.text:
  811. event_id = await self._handle_instagram_text(intent, item, item.text)
  812. elif item.like:
  813. # We handle likes as text because Matrix clients do big emoji on their own.
  814. event_id = await self._handle_instagram_text(intent, item, item.like)
  815. elif item.link:
  816. event_id = await self._handle_instagram_text(intent, item, item.link.text)
  817. handled = bool(event_id)
  818. if not event_id:
  819. self.log.debug(f"Unhandled Instagram message {item.item_id}")
  820. event_id = await self._send_instagram_unhandled(intent, item)
  821. msg = DBMessage(mxid=event_id, mx_room=self.mxid, item_id=item.item_id,
  822. receiver=self.receiver, sender=sender.pk)
  823. await msg.insert()
  824. await self._send_delivery_receipt(event_id)
  825. if handled:
  826. self.log.debug(f"Handled Instagram message {item.item_id} -> {event_id}")
  827. else:
  828. self.log.debug(f"Unhandled Instagram message {item.item_id} "
  829. f"(type {item.item_type} -> fallback error {event_id})")
  830. if is_backfill and item.reactions:
  831. await self._handle_instagram_reactions(msg, item.reactions.emojis)
  832. async def handle_instagram_remove(self, item_id: str) -> None:
  833. message = await DBMessage.get_by_item_id(item_id, self.receiver)
  834. if message is None:
  835. return
  836. await message.delete()
  837. sender = await p.Puppet.get_by_pk(message.sender)
  838. try:
  839. await sender.intent_for(self).redact(self.mxid, message.mxid)
  840. except MForbidden:
  841. await self.main_intent.redact(self.mxid, message.mxid)
  842. self.log.debug(f"Redacted {message.mxid} after Instagram unsend")
  843. async def _handle_instagram_reactions(self, message: DBMessage, reactions: List[Reaction]
  844. ) -> None:
  845. old_reactions: Dict[int, DBReaction]
  846. old_reactions = {reaction.ig_sender: reaction for reaction
  847. in await DBReaction.get_all_by_item_id(message.item_id, self.receiver)}
  848. for new_reaction in reactions:
  849. old_reaction = old_reactions.pop(new_reaction.sender_id, None)
  850. if old_reaction and old_reaction.reaction == new_reaction.emoji:
  851. continue
  852. puppet = await p.Puppet.get_by_pk(new_reaction.sender_id)
  853. intent = puppet.intent_for(self)
  854. reaction_event_id = await intent.react(self.mxid, message.mxid, new_reaction.emoji)
  855. await self._upsert_reaction(old_reaction, intent, reaction_event_id, message,
  856. puppet, new_reaction.emoji)
  857. for old_reaction in old_reactions.values():
  858. await old_reaction.delete()
  859. puppet = await p.Puppet.get_by_pk(old_reaction.ig_sender)
  860. await puppet.intent_for(self).redact(self.mxid, old_reaction.mxid)
  861. async def handle_instagram_update(self, item: MessageSyncMessage) -> None:
  862. message = await DBMessage.get_by_item_id(item.item_id, self.receiver)
  863. if not message:
  864. return
  865. if item.has_seen:
  866. puppet = await p.Puppet.get_by_pk(item.has_seen, create=False)
  867. if puppet:
  868. await puppet.intent_for(self).mark_read(self.mxid, message.mxid)
  869. else:
  870. async with self._reaction_lock:
  871. await self._handle_instagram_reactions(message, (item.reactions.emojis
  872. if item.reactions else []))
  873. # endregion
  874. # region Updating portal info
  875. def _get_thread_name(self, thread: Thread) -> str:
  876. if self.is_direct:
  877. if self.other_user_pk == thread.viewer_id and len(thread.users) == 0:
  878. return "Instagram chat with yourself"
  879. elif len(thread.users) == 1:
  880. tpl = self.config["bridge.private_chat_name_template"]
  881. ui = thread.users[0]
  882. return tpl.format(displayname=ui.full_name, id=ui.pk, username=ui.username)
  883. pass
  884. elif thread.thread_title:
  885. return self.config["bridge.group_chat_name_template"].format(name=thread.thread_title)
  886. else:
  887. return ""
  888. async def update_info(self, thread: Thread, source: 'u.User') -> None:
  889. changed = await self._update_name(self._get_thread_name(thread))
  890. changed = await self._update_participants(thread.users, source) or changed
  891. if changed:
  892. await self.update_bridge_info()
  893. await self.update()
  894. # TODO update power levels with thread.admin_user_ids
  895. async def _update_name(self, name: str) -> bool:
  896. if name and (self.name != name or not self.name_set):
  897. self.name = name
  898. if self.mxid:
  899. try:
  900. await self.main_intent.set_room_name(self.mxid, name)
  901. self.name_set = True
  902. except Exception:
  903. self.log.exception("Failed to update name")
  904. self.name_set = False
  905. return True
  906. return False
  907. async def _update_photo_from_puppet(self, puppet: 'p.Puppet') -> bool:
  908. if not self.private_chat_portal_meta and not self.encrypted:
  909. return False
  910. if self.avatar_set and self.avatar_url == puppet.photo_mxc:
  911. return False
  912. self.avatar_url = puppet.photo_mxc
  913. if self.mxid:
  914. try:
  915. await self.main_intent.set_room_avatar(self.mxid, puppet.photo_mxc)
  916. self.avatar_set = True
  917. except Exception:
  918. self.log.exception("Failed to set room avatar")
  919. self.avatar_set = False
  920. return True
  921. async def _update_participants(self, users: List[ThreadUser], source: 'u.User') -> bool:
  922. meta_changed = False
  923. # Make sure puppets who should be here are here
  924. for user in users:
  925. puppet = await p.Puppet.get_by_pk(user.pk)
  926. await puppet.update_info(user, source)
  927. if self.mxid:
  928. await puppet.intent_for(self).ensure_joined(self.mxid)
  929. if puppet.pk == self.other_user_pk:
  930. meta_changed = await self._update_photo_from_puppet(puppet)
  931. if self.mxid:
  932. # Kick puppets who shouldn't be here
  933. current_members = {int(user.pk) for user in users}
  934. for user_id in await self.main_intent.get_room_members(self.mxid):
  935. pk = p.Puppet.get_id_from_mxid(user_id)
  936. if pk and pk not in current_members and pk != self.other_user_pk:
  937. await self.main_intent.kick_user(self.mxid, p.Puppet.get_mxid_from_id(pk),
  938. reason="User had left this Instagram DM")
  939. return meta_changed
  940. async def _update_read_receipts(self, receipts: Dict[Union[int, str], ThreadUserLastSeenAt]
  941. ) -> None:
  942. for user_id, receipt in receipts.items():
  943. message = await DBMessage.get_by_item_id(receipt.item_id, self.receiver)
  944. if not message:
  945. continue
  946. puppet = await p.Puppet.get_by_pk(int(user_id), create=False)
  947. if not puppet:
  948. continue
  949. try:
  950. await puppet.intent_for(self).mark_read(message.mx_room, message.mxid)
  951. except Exception:
  952. self.log.warning(f"Failed to mark {message.mxid} in {message.mx_room} "
  953. f"as read by {puppet.intent.mxid}", exc_info=True)
  954. # endregion
  955. # region Backfilling
  956. async def backfill(self, source: 'u.User', is_initial: bool = False) -> None:
  957. limit = (self.config["bridge.backfill.initial_limit"] if is_initial
  958. else self.config["bridge.backfill.missed_limit"])
  959. if limit == 0:
  960. return
  961. elif limit < 0:
  962. limit = None
  963. with self.backfill_lock:
  964. await self._backfill(source, is_initial, limit)
  965. async def _backfill(self, source: 'u.User', is_initial: bool, limit: int) -> None:
  966. self.log.debug("Backfilling history through %s", source.mxid)
  967. entries = await self._fetch_backfill_items(source, is_initial, limit)
  968. if not entries:
  969. self.log.debug("Didn't get any items to backfill from server")
  970. return
  971. self.log.debug("Got %d entries from server", len(entries))
  972. self._backfill_leave = set()
  973. async with NotificationDisabler(self.mxid, source):
  974. for entry in reversed(entries):
  975. sender = await p.Puppet.get_by_pk(int(entry.user_id))
  976. await self.handle_instagram_item(source, sender, entry, is_backfill=True)
  977. for intent in self._backfill_leave:
  978. self.log.trace("Leaving room with %s post-backfill", intent.mxid)
  979. await intent.leave_room(self.mxid)
  980. self._backfill_leave = None
  981. self.log.info("Backfilled %d messages through %s", len(entries), source.mxid)
  982. async def _fetch_backfill_items(self, source: 'u.User', is_initial: bool, limit: int
  983. ) -> List[ThreadItem]:
  984. items = []
  985. self.log.debug("Fetching up to %d messages through %s", limit, source.igpk)
  986. async for item in source.client.iter_thread(self.thread_id):
  987. if len(items) >= limit:
  988. self.log.debug(f"Fetched {len(items)} messages (the limit)")
  989. break
  990. elif not is_initial:
  991. msg = await DBMessage.get_by_item_id(item.item_id, receiver=self.receiver)
  992. if msg is not None:
  993. self.log.debug(f"Fetched {len(items)} messages and hit a message"
  994. " that's already in the database.")
  995. break
  996. items.append(item)
  997. return items
  998. # endregion
  999. # region Bridge info state event
  1000. @property
  1001. def bridge_info_state_key(self) -> str:
  1002. return f"net.maunium.instagram://instagram/{self.thread_id}"
  1003. @property
  1004. def bridge_info(self) -> Dict[str, Any]:
  1005. return {
  1006. "bridgebot": self.az.bot_mxid,
  1007. "creator": self.main_intent.mxid,
  1008. "protocol": {
  1009. "id": "instagram",
  1010. "displayname": "Instagram DM",
  1011. "avatar_url": self.config["appservice.bot_avatar"],
  1012. },
  1013. "channel": {
  1014. "id": self.thread_id,
  1015. "displayname": self.name,
  1016. "avatar_url": self.avatar_url,
  1017. }
  1018. }
  1019. async def update_bridge_info(self) -> None:
  1020. if not self.mxid:
  1021. self.log.debug("Not updating bridge info: no Matrix room created")
  1022. return
  1023. try:
  1024. self.log.debug("Updating bridge info...")
  1025. await self.main_intent.send_state_event(self.mxid, StateBridge,
  1026. self.bridge_info, self.bridge_info_state_key)
  1027. # TODO remove this once https://github.com/matrix-org/matrix-doc/pull/2346 is in spec
  1028. await self.main_intent.send_state_event(self.mxid, StateHalfShotBridge,
  1029. self.bridge_info, self.bridge_info_state_key)
  1030. except Exception:
  1031. self.log.warning("Failed to update bridge info", exc_info=True)
  1032. # endregion
  1033. # region Creating Matrix rooms
  1034. async def create_matrix_room(self, source: 'u.User', info: Thread) -> Optional[RoomID]:
  1035. if self.mxid:
  1036. try:
  1037. await self.update_matrix_room(source, info)
  1038. except Exception:
  1039. self.log.exception("Failed to update portal")
  1040. return self.mxid
  1041. async with self._create_room_lock:
  1042. return await self._create_matrix_room(source, info)
  1043. def _get_invite_content(self, double_puppet: Optional['p.Puppet']) -> Dict[str, Any]:
  1044. invite_content = {}
  1045. if double_puppet:
  1046. invite_content["fi.mau.will_auto_accept"] = True
  1047. if self.is_direct:
  1048. invite_content["is_direct"] = True
  1049. return invite_content
  1050. async def update_matrix_room(self, source: 'u.User', info: Thread, backfill: bool = False
  1051. ) -> None:
  1052. puppet = await p.Puppet.get_by_custom_mxid(source.mxid)
  1053. await self.main_intent.invite_user(self.mxid, source.mxid, check_cache=True,
  1054. extra_content=self._get_invite_content(puppet))
  1055. if puppet:
  1056. did_join = await puppet.intent.ensure_joined(self.mxid)
  1057. if did_join and self.is_direct:
  1058. await source.update_direct_chats({self.main_intent.mxid: [self.mxid]})
  1059. await self.update_info(info, source)
  1060. if backfill:
  1061. last_msg = await DBMessage.get_by_item_id(info.last_permanent_item.item_id,
  1062. receiver=self.receiver)
  1063. if last_msg is None:
  1064. self.log.debug(f"Last permanent item ({info.last_permanent_item.item_id})"
  1065. " not found in database, starting backfilling")
  1066. await self.backfill(source, is_initial=False)
  1067. await self._update_read_receipts(info.last_seen_at)
  1068. # TODO
  1069. # up = DBUserPortal.get(source.fbid, self.fbid, self.fb_receiver)
  1070. # if not up:
  1071. # in_community = await source._community_helper.add_room(source._community_id, self.mxid)
  1072. # DBUserPortal(user=source.fbid, portal=self.fbid, portal_receiver=self.fb_receiver,
  1073. # in_community=in_community).insert()
  1074. # elif not up.in_community:
  1075. # in_community = await source._community_helper.add_room(source._community_id, self.mxid)
  1076. # up.edit(in_community=in_community)
  1077. async def _create_matrix_room(self, source: 'u.User', info: Thread) -> Optional[RoomID]:
  1078. if self.mxid:
  1079. await self.update_matrix_room(source, info)
  1080. return self.mxid
  1081. await self.update_info(info, source)
  1082. self.log.debug("Creating Matrix room")
  1083. name: Optional[str] = None
  1084. initial_state = [{
  1085. "type": str(StateBridge),
  1086. "state_key": self.bridge_info_state_key,
  1087. "content": self.bridge_info,
  1088. }, {
  1089. # TODO remove this once https://github.com/matrix-org/matrix-doc/pull/2346 is in spec
  1090. "type": str(StateHalfShotBridge),
  1091. "state_key": self.bridge_info_state_key,
  1092. "content": self.bridge_info,
  1093. }]
  1094. invites = []
  1095. if self.config["bridge.encryption.default"] and self.matrix.e2ee:
  1096. self.encrypted = True
  1097. initial_state.append({
  1098. "type": "m.room.encryption",
  1099. "content": {"algorithm": "m.megolm.v1.aes-sha2"},
  1100. })
  1101. if self.is_direct:
  1102. invites.append(self.az.bot_mxid)
  1103. if self.encrypted or self.private_chat_portal_meta or not self.is_direct:
  1104. name = self.name
  1105. if self.config["appservice.community_id"]:
  1106. initial_state.append({
  1107. "type": "m.room.related_groups",
  1108. "content": {"groups": [self.config["appservice.community_id"]]},
  1109. })
  1110. # We lock backfill lock here so any messages that come between the room being created
  1111. # and the initial backfill finishing wouldn't be bridged before the backfill messages.
  1112. with self.backfill_lock:
  1113. creation_content = {}
  1114. if not self.config["bridge.federate_rooms"]:
  1115. creation_content["m.federate"] = False
  1116. self.mxid = await self.main_intent.create_room(
  1117. name=name,
  1118. is_direct=self.is_direct,
  1119. initial_state=initial_state,
  1120. invitees=invites,
  1121. creation_content=creation_content,
  1122. )
  1123. if not self.mxid:
  1124. raise Exception("Failed to create room: no mxid returned")
  1125. if self.encrypted and self.matrix.e2ee and self.is_direct:
  1126. try:
  1127. await self.az.intent.ensure_joined(self.mxid)
  1128. except Exception:
  1129. self.log.warning("Failed to add bridge bot "
  1130. f"to new private chat {self.mxid}")
  1131. await self.update()
  1132. self.log.debug(f"Matrix room created: {self.mxid}")
  1133. self.by_mxid[self.mxid] = self
  1134. puppet = await p.Puppet.get_by_custom_mxid(source.mxid)
  1135. await self.main_intent.invite_user(self.mxid, source.mxid,
  1136. extra_content=self._get_invite_content(puppet))
  1137. if puppet:
  1138. try:
  1139. if self.is_direct:
  1140. await source.update_direct_chats({self.main_intent.mxid: [self.mxid]})
  1141. await puppet.intent.join_room_by_id(self.mxid)
  1142. except MatrixError:
  1143. self.log.debug("Failed to join custom puppet into newly created portal",
  1144. exc_info=True)
  1145. await self._update_participants(info.users, source)
  1146. # TODO
  1147. # in_community = await source._community_helper.add_room(source._community_id, self.mxid)
  1148. # DBUserPortal(user=source.fbid, portal=self.fbid, portal_receiver=self.fb_receiver,
  1149. # in_community=in_community).upsert()
  1150. try:
  1151. await self.backfill(source, is_initial=True)
  1152. except Exception:
  1153. self.log.exception("Failed to backfill new portal")
  1154. await self._update_read_receipts(info.last_seen_at)
  1155. return self.mxid
  1156. # endregion
  1157. # region Database getters
  1158. async def postinit(self) -> None:
  1159. self.by_thread_id[(self.thread_id, self.receiver)] = self
  1160. if self.mxid:
  1161. self.by_mxid[self.mxid] = self
  1162. self._main_intent = ((await p.Puppet.get_by_pk(self.other_user_pk)).default_mxid_intent
  1163. if self.other_user_pk else self.az.intent)
  1164. async def delete(self) -> None:
  1165. await DBMessage.delete_all(self.mxid)
  1166. self.by_mxid.pop(self.mxid, None)
  1167. self.mxid = None
  1168. self.encrypted = False
  1169. await self.update()
  1170. async def save(self) -> None:
  1171. await self.update()
  1172. @classmethod
  1173. def all_with_room(cls) -> AsyncGenerator['Portal', None]:
  1174. return cls._db_to_portals(super().all_with_room())
  1175. @classmethod
  1176. def find_private_chats_with(cls, other_user: int) -> AsyncGenerator['Portal', None]:
  1177. return cls._db_to_portals(super().find_private_chats_with(other_user))
  1178. @classmethod
  1179. async def _db_to_portals(cls, query: Awaitable[List['Portal']]
  1180. ) -> AsyncGenerator['Portal', None]:
  1181. portals = await query
  1182. for index, portal in enumerate(portals):
  1183. try:
  1184. yield cls.by_thread_id[(portal.thread_id, portal.receiver)]
  1185. except KeyError:
  1186. await portal.postinit()
  1187. yield portal
  1188. @classmethod
  1189. @async_getter_lock
  1190. async def get_by_mxid(cls, mxid: RoomID) -> Optional['Portal']:
  1191. try:
  1192. return cls.by_mxid[mxid]
  1193. except KeyError:
  1194. pass
  1195. portal = cast(cls, await super().get_by_mxid(mxid))
  1196. if portal is not None:
  1197. await portal.postinit()
  1198. return portal
  1199. return None
  1200. @classmethod
  1201. @async_getter_lock
  1202. async def get_by_thread_id(cls, thread_id: str, *, receiver: int,
  1203. is_group: Optional[bool] = None,
  1204. other_user_pk: Optional[int] = None) -> Optional['Portal']:
  1205. if is_group and receiver != 0:
  1206. receiver = 0
  1207. try:
  1208. return cls.by_thread_id[(thread_id, receiver)]
  1209. except KeyError:
  1210. pass
  1211. if is_group is None and receiver != 0:
  1212. try:
  1213. return cls.by_thread_id[(thread_id, 0)]
  1214. except KeyError:
  1215. pass
  1216. portal = cast(cls, await super().get_by_thread_id(thread_id, receiver=receiver,
  1217. rec_must_match=is_group is not None))
  1218. if portal is not None:
  1219. await portal.postinit()
  1220. return portal
  1221. if is_group is not None:
  1222. portal = cls(thread_id, receiver, other_user_pk=other_user_pk)
  1223. await portal.insert()
  1224. await portal.postinit()
  1225. return portal
  1226. return None
  1227. @classmethod
  1228. async def get_by_thread(cls, thread: Thread, receiver: int) -> Optional['Portal']:
  1229. if thread.is_group:
  1230. receiver = 0
  1231. other_user_pk = None
  1232. else:
  1233. if len(thread.users) == 0:
  1234. other_user_pk = receiver
  1235. else:
  1236. other_user_pk = thread.users[0].pk
  1237. return await cls.get_by_thread_id(thread.thread_id, receiver=receiver,
  1238. is_group=thread.is_group, other_user_pk=other_user_pk)
  1239. # endregion