portal.py 103 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610
  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, Optional, Union, cast
  18. from collections import deque
  19. from io import BytesIO
  20. import asyncio
  21. import base64
  22. import hashlib
  23. import html
  24. import json
  25. import mimetypes
  26. import re
  27. import sqlite3
  28. import time
  29. from yarl import URL
  30. import asyncpg
  31. import magic
  32. from mauigpapi.errors import IGRateLimitError
  33. from mauigpapi.types import (
  34. AnimatedMediaItem,
  35. CommandResponse,
  36. ExpiredMediaItem,
  37. MediaShareItem,
  38. MediaType,
  39. MessageSyncMessage,
  40. Reaction,
  41. ReactionStatus,
  42. ReelMediaShareItem,
  43. ReelShareType,
  44. RegularMediaItem,
  45. Thread,
  46. ThreadImageCandidate,
  47. ThreadItem,
  48. ThreadItemType,
  49. ThreadUser,
  50. ThreadUserLastSeenAt,
  51. TypingStatus,
  52. VoiceMediaItem,
  53. XMAMediaShareItem,
  54. )
  55. from mautrix.appservice import DOUBLE_PUPPET_SOURCE_KEY, IntentAPI
  56. from mautrix.bridge import BasePortal, async_getter_lock
  57. from mautrix.errors import MatrixError, MForbidden, MNotFound, SessionNotFound
  58. from mautrix.types import (
  59. AudioInfo,
  60. BatchID,
  61. BatchSendEvent,
  62. BatchSendStateEvent,
  63. BeeperMessageStatusEventContent,
  64. ContentURI,
  65. EventID,
  66. EventType,
  67. Format,
  68. ImageInfo,
  69. LocationMessageEventContent,
  70. MediaMessageEventContent,
  71. Membership,
  72. MemberStateEventContent,
  73. MessageEventContent,
  74. MessageStatus,
  75. MessageStatusReason,
  76. MessageType,
  77. ReactionEventContent,
  78. RelatesTo,
  79. RelationType,
  80. RoomID,
  81. TextMessageEventContent,
  82. UserID,
  83. VideoInfo,
  84. )
  85. from mautrix.util import ffmpeg
  86. from mautrix.util.message_send_checkpoint import MessageSendCheckpointStatus
  87. from . import matrix as m, puppet as p, user as u
  88. from .config import Config
  89. from .db import Backfill, Message as DBMessage, Portal as DBPortal, Reaction as DBReaction
  90. if TYPE_CHECKING:
  91. from .__main__ import InstagramBridge
  92. try:
  93. from mautrix.crypto.attachments import decrypt_attachment, encrypt_attachment
  94. except ImportError:
  95. encrypt_attachment = decrypt_attachment = None
  96. try:
  97. from PIL import Image
  98. except ImportError:
  99. Image = None
  100. StateBridge = EventType.find("m.bridge", EventType.Class.STATE)
  101. StateHalfShotBridge = EventType.find("uk.half-shot.bridge", EventType.Class.STATE)
  102. MediaData = Union[
  103. AnimatedMediaItem,
  104. ExpiredMediaItem,
  105. MediaShareItem,
  106. ReelMediaShareItem,
  107. RegularMediaItem,
  108. VoiceMediaItem,
  109. XMAMediaShareItem,
  110. ]
  111. MediaUploadFunc = Callable[["u.User", MediaData, IntentAPI], Awaitable[MediaMessageEventContent]]
  112. PortalCreateDummy = EventType.find("fi.mau.dummy.portal_created", EventType.Class.MESSAGE)
  113. HistorySyncMarkerMessage = EventType.find("org.matrix.msc2716.marker", EventType.Class.MESSAGE)
  114. ConvertedMessage = tuple[EventType, MessageEventContent]
  115. # This doesn't need to capture all valid URLs, it's enough to catch most of them.
  116. # False negatives simply mean the link won't be linkified on Instagram,
  117. # but false positives will cause the message to fail to send.
  118. SIMPLE_URL_REGEX = re.compile(
  119. r"(?P<url>https?://[\da-z.-]+\.[a-z]{2,}(?:/[^\s]*)?)", flags=re.IGNORECASE
  120. )
  121. class UnsupportedAttachmentError(NotImplementedError):
  122. pass
  123. class Portal(DBPortal, BasePortal):
  124. by_mxid: dict[RoomID, Portal] = {}
  125. by_thread_id: dict[tuple[str, int], Portal] = {}
  126. config: Config
  127. matrix: m.MatrixHandler
  128. private_chat_portal_meta: bool
  129. _main_intent: IntentAPI | None
  130. _create_room_lock: asyncio.Lock
  131. _msgid_dedup: deque[str]
  132. _reqid_dedup: set[str]
  133. _last_participant_update: set[int]
  134. _reaction_lock: asyncio.Lock
  135. _typing: set[UserID]
  136. def __init__(
  137. self,
  138. thread_id: str,
  139. receiver: int,
  140. other_user_pk: int | None,
  141. mxid: RoomID | None = None,
  142. name: str | None = None,
  143. avatar_url: ContentURI | None = None,
  144. encrypted: bool = False,
  145. name_set: bool = False,
  146. avatar_set: bool = False,
  147. relay_user_id: UserID | None = None,
  148. first_event_id: EventID | None = None,
  149. next_batch_id: BatchID | None = None,
  150. historical_base_insertion_event_id: EventID | None = None,
  151. cursor: str | None = None,
  152. thread_image_id: int | None = None,
  153. ) -> None:
  154. super().__init__(
  155. thread_id,
  156. receiver,
  157. other_user_pk,
  158. mxid,
  159. name,
  160. avatar_url,
  161. encrypted,
  162. name_set,
  163. avatar_set,
  164. relay_user_id,
  165. first_event_id,
  166. next_batch_id,
  167. historical_base_insertion_event_id,
  168. cursor,
  169. thread_image_id,
  170. )
  171. self._create_room_lock = asyncio.Lock()
  172. self.log = self.log.getChild(thread_id)
  173. self._msgid_dedup = deque(maxlen=100)
  174. self._reqid_dedup = set()
  175. self._last_participant_update = set()
  176. self._main_intent = None
  177. self._reaction_lock = asyncio.Lock()
  178. self._typing = set()
  179. self._relay_user = None
  180. @property
  181. def is_direct(self) -> bool:
  182. return self.other_user_pk is not None
  183. @property
  184. def main_intent(self) -> IntentAPI:
  185. if not self._main_intent:
  186. raise ValueError("Portal must be postinit()ed before main_intent can be used")
  187. return self._main_intent
  188. @classmethod
  189. def init_cls(cls, bridge: "InstagramBridge") -> None:
  190. BasePortal.bridge = bridge
  191. cls.config = bridge.config
  192. cls.matrix = bridge.matrix
  193. cls.az = bridge.az
  194. cls.loop = bridge.loop
  195. cls.bridge = bridge
  196. cls.private_chat_portal_meta = cls.config["bridge.private_chat_portal_meta"]
  197. # region Misc
  198. async def _send_delivery_receipt(self, event_id: EventID) -> None:
  199. if event_id and self.config["bridge.delivery_receipts"]:
  200. try:
  201. await self.az.intent.mark_read(self.mxid, event_id)
  202. except Exception:
  203. self.log.exception("Failed to send delivery receipt for %s", event_id)
  204. async def _send_bridge_success(
  205. self,
  206. sender: u.User,
  207. event_id: EventID,
  208. event_type: EventType,
  209. msgtype: MessageType | None = None,
  210. ) -> None:
  211. sender.send_remote_checkpoint(
  212. status=MessageSendCheckpointStatus.SUCCESS,
  213. event_id=event_id,
  214. room_id=self.mxid,
  215. event_type=event_type,
  216. message_type=msgtype,
  217. )
  218. asyncio.create_task(self._send_message_status(event_id, err=None))
  219. await self._send_delivery_receipt(event_id)
  220. async def _send_bridge_error(
  221. self,
  222. sender: u.User,
  223. err: Exception,
  224. event_id: EventID,
  225. event_type: EventType,
  226. message_type: MessageType | None = None,
  227. confirmed: bool = False,
  228. ) -> None:
  229. sender.send_remote_checkpoint(
  230. self._status_from_exception(err),
  231. event_id,
  232. self.mxid,
  233. event_type,
  234. message_type=message_type,
  235. error=err,
  236. )
  237. if self.config["bridge.delivery_error_reports"]:
  238. event_type_str = {
  239. EventType.REACTION: "reaction",
  240. EventType.ROOM_REDACTION: "redaction",
  241. }.get(event_type, "message")
  242. error_type = "was not" if confirmed else "may not have been"
  243. await self._send_message(
  244. self.main_intent,
  245. TextMessageEventContent(
  246. msgtype=MessageType.NOTICE,
  247. body=f"\u26a0 Your {event_type_str} {error_type} bridged: {str(err)}",
  248. ),
  249. )
  250. asyncio.create_task(self._send_message_status(event_id, err))
  251. async def _send_message_status(self, event_id: EventID, err: Exception | None) -> None:
  252. if not self.config["bridge.message_status_events"]:
  253. return
  254. intent = self.az.intent if self.encrypted else self.main_intent
  255. status = BeeperMessageStatusEventContent(
  256. network=self.bridge_info_state_key,
  257. relates_to=RelatesTo(
  258. rel_type=RelationType.REFERENCE,
  259. event_id=event_id,
  260. ),
  261. )
  262. if err:
  263. status.error = str(err)
  264. if isinstance(err, NotImplementedError):
  265. if isinstance(err, UnsupportedAttachmentError):
  266. status.message = str(err)
  267. status.reason = MessageStatusReason.UNSUPPORTED
  268. status.status = MessageStatus.FAIL
  269. else:
  270. status.reason = MessageStatusReason.GENERIC_ERROR
  271. status.status = MessageStatus.RETRIABLE
  272. else:
  273. status.status = MessageStatus.SUCCESS
  274. status.fill_legacy_booleans()
  275. await intent.send_message_event(
  276. room_id=self.mxid,
  277. event_type=EventType.BEEPER_MESSAGE_STATUS,
  278. content=status,
  279. )
  280. async def _upsert_reaction(
  281. self,
  282. existing: DBReaction | None,
  283. intent: IntentAPI,
  284. mxid: EventID,
  285. message: DBMessage,
  286. sender: u.User | p.Puppet,
  287. reaction: str,
  288. mx_timestamp: int,
  289. ) -> None:
  290. if existing:
  291. self.log.debug(
  292. f"_upsert_reaction redacting {existing.mxid} and inserting {mxid}"
  293. f" (message: {message.mxid})"
  294. )
  295. await intent.redact(existing.mx_room, existing.mxid)
  296. await existing.edit(
  297. reaction=reaction, mxid=mxid, mx_room=message.mx_room, mx_timestamp=mx_timestamp
  298. )
  299. else:
  300. self.log.debug(f"_upsert_reaction inserting {mxid} (message: {message.mxid})")
  301. await DBReaction(
  302. mxid=mxid,
  303. mx_room=message.mx_room,
  304. ig_item_id=message.item_id,
  305. ig_receiver=self.receiver,
  306. ig_sender=sender.igpk,
  307. reaction=reaction,
  308. mx_timestamp=mx_timestamp,
  309. ).insert()
  310. # endregion
  311. # region Matrix event handling
  312. @staticmethod
  313. def _status_from_exception(e: Exception) -> MessageSendCheckpointStatus:
  314. if isinstance(e, NotImplementedError):
  315. return MessageSendCheckpointStatus.UNSUPPORTED
  316. elif isinstance(e, asyncio.TimeoutError):
  317. return MessageSendCheckpointStatus.TIMEOUT
  318. return MessageSendCheckpointStatus.PERM_FAILURE
  319. async def handle_matrix_message(
  320. self, sender: u.User, message: MessageEventContent, event_id: EventID
  321. ) -> None:
  322. try:
  323. await self._handle_matrix_message(sender, message, event_id)
  324. except Exception as e:
  325. self.log.exception(f"Error handling Matrix event {event_id}")
  326. await self._send_bridge_error(
  327. sender,
  328. e,
  329. event_id,
  330. EventType.ROOM_MESSAGE,
  331. message_type=message.msgtype,
  332. confirmed=True,
  333. )
  334. else:
  335. await self._send_bridge_success(
  336. sender, event_id, EventType.ROOM_MESSAGE, message.msgtype
  337. )
  338. async def _handle_matrix_giphy(
  339. self,
  340. sender: u.User,
  341. event_id: EventID,
  342. request_id: str,
  343. giphy_id: str,
  344. ) -> CommandResponse:
  345. self.log.trace(f"Broadcasting giphy from {event_id} with request ID {request_id}")
  346. return await sender.client.broadcast(
  347. self.thread_id,
  348. ThreadItemType.ANIMATED_MEDIA,
  349. client_context=request_id,
  350. id=giphy_id,
  351. )
  352. async def _handle_matrix_image(
  353. self,
  354. sender: u.User,
  355. event_id: EventID,
  356. request_id: str,
  357. data: bytes,
  358. mime_type: str,
  359. width: int | None = None,
  360. height: int | None = None,
  361. ) -> CommandResponse:
  362. if mime_type != "image/jpeg":
  363. if Image is None:
  364. raise UnsupportedAttachmentError(
  365. "Instagram does not allow non-JPEG images, and Pillow is not installed, "
  366. "so the bridge couldn't convert the image automatically"
  367. )
  368. with BytesIO(data) as inp, BytesIO() as out:
  369. img = Image.open(inp)
  370. img.convert("RGB").save(out, format="JPEG", quality=80)
  371. data = out.getvalue()
  372. mime_type = "image/jpeg"
  373. self.log.trace(f"Uploading photo from {event_id} (mime: {mime_type})")
  374. upload_resp = await sender.client.upload_photo(
  375. data, mime=mime_type, width=width, height=height
  376. )
  377. self.log.trace(f"Broadcasting uploaded photo with request ID {request_id}")
  378. return await sender.client.broadcast(
  379. self.thread_id,
  380. ThreadItemType.CONFIGURE_PHOTO,
  381. client_context=request_id,
  382. upload_id=upload_resp.upload_id,
  383. allow_full_aspect_ratio="true",
  384. )
  385. async def _handle_matrix_video(
  386. self,
  387. sender: u.User,
  388. event_id: EventID,
  389. request_id: str,
  390. data: bytes,
  391. mime_type: str,
  392. duration: int | None = None,
  393. width: int | None = None,
  394. height: int | None = None,
  395. ) -> CommandResponse:
  396. if mime_type != "video/mp4":
  397. data = await ffmpeg.convert_bytes(
  398. data,
  399. output_extension=".mp4",
  400. output_args=("-c:v", "libx264", "-c:a", "aac"),
  401. input_mime=mime_type,
  402. )
  403. self.log.trace(f"Uploading video from {event_id}")
  404. _, upload_id = await sender.client.upload_mp4(
  405. data, duration_ms=duration, width=width, height=height
  406. )
  407. self.log.trace(f"Broadcasting uploaded video with request ID {request_id}")
  408. return await sender.client.broadcast(
  409. self.thread_id,
  410. ThreadItemType.CONFIGURE_VIDEO,
  411. client_context=request_id,
  412. upload_id=upload_id,
  413. video_result="",
  414. )
  415. async def _handle_matrix_audio(
  416. self,
  417. sender: u.User,
  418. event_id: EventID,
  419. request_id: str,
  420. data: bytes,
  421. mime_type: str,
  422. waveform: list[int],
  423. duration: int | None = None,
  424. ) -> CommandResponse:
  425. if mime_type != "audio/mp4":
  426. data = await ffmpeg.convert_bytes(
  427. data, output_extension=".m4a", output_args=("-c:a", "aac"), input_mime=mime_type
  428. )
  429. self.log.trace(f"Uploading audio from {event_id}")
  430. _, upload_id = await sender.client.upload_mp4(data, audio=True, duration_ms=duration)
  431. self.log.trace(f"Broadcasting uploaded audio with request ID {request_id}")
  432. return await sender.client.broadcast(
  433. self.thread_id,
  434. ThreadItemType.SHARE_VOICE,
  435. client_context=request_id,
  436. upload_id=upload_id,
  437. waveform=json.dumps([(part or 0) / 1024 for part in waveform], separators=(",", ":")),
  438. waveform_sampling_frequency_hz="10",
  439. )
  440. async def _handle_matrix_message(
  441. self, orig_sender: u.User, message: MessageEventContent, event_id: EventID
  442. ) -> None:
  443. sender, is_relay = await self.get_relay_sender(orig_sender, f"message {event_id}")
  444. assert sender, "user is not logged in"
  445. assert sender.is_connected, "You're not connected to Instagram"
  446. if is_relay:
  447. await self.apply_relay_message_format(orig_sender, message)
  448. reply_to = {}
  449. if message.get_reply_to():
  450. msg = await DBMessage.get_by_mxid(message.get_reply_to(), self.mxid)
  451. if msg and msg.client_context:
  452. reply_to = {
  453. "replied_to_item_id": msg.item_id,
  454. "replied_to_client_context": msg.client_context,
  455. }
  456. request_id = sender.state.gen_client_context()
  457. self._reqid_dedup.add(request_id)
  458. self.log.debug(
  459. f"Handling Matrix message {event_id} from {sender.mxid}/{sender.igpk} "
  460. f"with request ID {request_id}"
  461. )
  462. if message.msgtype == MessageType.NOTICE and not self.config["bridge.bridge_notices"]:
  463. return
  464. if message.msgtype in (MessageType.EMOTE, MessageType.TEXT, MessageType.NOTICE):
  465. text = message.body
  466. if message.msgtype == MessageType.EMOTE:
  467. text = f"/me {text}"
  468. self.log.trace(f"Sending Matrix text from {event_id} with request ID {request_id}")
  469. urls = SIMPLE_URL_REGEX.findall(text) or None
  470. if not self.is_direct:
  471. # Instagram groups don't seem to support sending link previews,
  472. # and the client_context-based deduplication breaks when trying to send them.
  473. urls = None
  474. resp = await sender.mqtt.send_text(
  475. self.thread_id, text=text, urls=urls, client_context=request_id, **reply_to
  476. )
  477. elif message.msgtype.is_media and "fi.mau.instagram.giphy_id" in message:
  478. resp = await self._handle_matrix_giphy(
  479. sender, event_id, request_id, message["fi.mau.instagram.giphy_id"]
  480. )
  481. elif message.msgtype.is_media:
  482. if message.file and decrypt_attachment:
  483. data = await self.main_intent.download_media(message.file.url)
  484. data = decrypt_attachment(
  485. data, message.file.key.key, message.file.hashes.get("sha256"), message.file.iv
  486. )
  487. else:
  488. data = await self.main_intent.download_media(message.url)
  489. mime_type = message.info.mimetype or magic.from_buffer(data, mime=True)
  490. if message.msgtype == MessageType.IMAGE:
  491. resp = await self._handle_matrix_image(
  492. sender,
  493. event_id,
  494. request_id,
  495. data,
  496. mime_type,
  497. width=message.info.width,
  498. height=message.info.height,
  499. )
  500. elif message.msgtype == MessageType.AUDIO:
  501. waveform = message.get("org.matrix.msc1767.audio", {}).get("waveform", [0] * 30)
  502. resp = await self._handle_matrix_audio(
  503. sender,
  504. event_id,
  505. request_id,
  506. data,
  507. mime_type,
  508. waveform,
  509. duration=message.info.duration,
  510. )
  511. elif message.msgtype == MessageType.VIDEO:
  512. resp = await self._handle_matrix_video(
  513. sender,
  514. event_id,
  515. request_id,
  516. data,
  517. mime_type,
  518. duration=message.info.duration,
  519. width=message.info.width,
  520. height=message.info.height,
  521. )
  522. else:
  523. raise UnsupportedAttachmentError(
  524. "Non-image/video/audio files are currently not supported"
  525. )
  526. else:
  527. raise NotImplementedError(f"Unknown message type {message.msgtype}")
  528. self.log.trace(f"Got response to message send {request_id}: {resp}")
  529. if resp.status != "ok" or not resp.payload:
  530. self.log.warning(f"Failed to handle {event_id}: {resp}")
  531. raise Exception(f"Sending message failed: {resp.error_message}")
  532. else:
  533. self._msgid_dedup.appendleft(resp.payload.item_id)
  534. try:
  535. await DBMessage(
  536. mxid=event_id,
  537. mx_room=self.mxid,
  538. item_id=resp.payload.item_id,
  539. client_context=resp.payload.client_context,
  540. receiver=self.receiver,
  541. sender=sender.igpk,
  542. ig_timestamp=int(resp.payload.timestamp),
  543. ).insert()
  544. except (asyncpg.UniqueViolationError, sqlite3.IntegrityError) as e:
  545. self.log.warning(
  546. f"Error while persisting {event_id} ({resp.payload.client_context}) "
  547. f"-> {resp.payload.item_id}: {e}"
  548. )
  549. self._reqid_dedup.remove(request_id)
  550. self.log.debug(
  551. f"Handled Matrix message {event_id} ({resp.payload.client_context}) "
  552. f"-> {resp.payload.item_id}"
  553. )
  554. async def handle_matrix_reaction(
  555. self, sender: u.User, event_id: EventID, reacting_to: EventID, emoji: str, timestamp: int
  556. ) -> None:
  557. try:
  558. await self._handle_matrix_reaction(sender, event_id, reacting_to, emoji, timestamp)
  559. except Exception as e:
  560. self.log.exception(f"Error handling Matrix reaction {event_id}")
  561. await self._send_bridge_error(
  562. sender,
  563. e,
  564. event_id,
  565. EventType.REACTION,
  566. confirmed=True,
  567. )
  568. else:
  569. await self._send_bridge_success(sender, event_id, EventType.REACTION)
  570. async def _handle_matrix_reaction(
  571. self, sender: u.User, event_id: EventID, reacting_to: EventID, emoji: str, timestamp: int
  572. ) -> None:
  573. if not await sender.is_logged_in():
  574. self.log.debug(f"Ignoring reaction by non-logged-in user {sender.mxid}")
  575. raise NotImplementedError("User is not logged in")
  576. message = await DBMessage.get_by_mxid(reacting_to, self.mxid)
  577. if not message or message.is_internal:
  578. self.log.debug(f"Ignoring reaction to unknown event {reacting_to}")
  579. await self.main_intent.redact(self.mxid, event_id, reason="Unknown target message")
  580. raise NotImplementedError("Unknown target message")
  581. existing = await DBReaction.get_by_item_id(message.item_id, message.receiver, sender.igpk)
  582. if existing and existing.reaction == emoji:
  583. return
  584. async with self._reaction_lock:
  585. resp = await sender.mqtt.send_reaction(
  586. self.thread_id, item_id=message.item_id, emoji=emoji
  587. )
  588. if resp.status != "ok":
  589. if resp.payload and resp.payload.message == "invalid unicode emoji":
  590. # Instagram doesn't support this reaction. Notify the user, and redact it
  591. # so that it doesn't get confusing.
  592. await self.main_intent.redact(self.mxid, event_id, reason="Unsupported emoji")
  593. raise NotImplementedError(f"Instagram does not support the {emoji} emoji.")
  594. raise Exception(f"Unknown response error: {resp}")
  595. self.log.trace(f"{sender.mxid} reacted to {message.item_id} with {emoji}")
  596. await self._upsert_reaction(
  597. existing, self.main_intent, event_id, message, sender, emoji, timestamp
  598. )
  599. async def handle_matrix_redaction(
  600. self, orig_sender: u.User, event_id: EventID, redaction_event_id: EventID
  601. ) -> None:
  602. sender = None
  603. try:
  604. sender, _ = await self.get_relay_sender(orig_sender, f"redaction {event_id}")
  605. if not sender:
  606. raise Exception("User is not logged in")
  607. await self._handle_matrix_redaction(sender, event_id)
  608. except Exception as e:
  609. self.log.exception(f"Error handling Matrix redaction {event_id}")
  610. await self._send_bridge_error(
  611. sender or orig_sender,
  612. e,
  613. redaction_event_id,
  614. EventType.ROOM_REDACTION,
  615. confirmed=True,
  616. )
  617. else:
  618. await self._send_bridge_success(sender, redaction_event_id, EventType.ROOM_REDACTION)
  619. async def _handle_matrix_redaction(self, sender: u.User, event_id: EventID) -> None:
  620. if not sender.is_connected:
  621. raise Exception("You're not connected to Instagram")
  622. reaction = await DBReaction.get_by_mxid(event_id, self.mxid)
  623. if reaction:
  624. try:
  625. await reaction.delete()
  626. await sender.mqtt.send_reaction(
  627. self.thread_id,
  628. item_id=reaction.ig_item_id,
  629. reaction_status=ReactionStatus.DELETED,
  630. emoji="",
  631. )
  632. except Exception as e:
  633. raise Exception(f"Removing reaction failed: {e}")
  634. else:
  635. self.log.trace(f"Removed reaction to {reaction.ig_item_id} after Matrix redaction")
  636. return
  637. message = await DBMessage.get_by_mxid(event_id, self.mxid)
  638. if message and not message.is_internal:
  639. try:
  640. await message.delete()
  641. await sender.client.delete_item(self.thread_id, message.item_id)
  642. self.log.trace(f"Removed {message} after Matrix redaction")
  643. except Exception as e:
  644. raise Exception(f"Removing message failed: {e}")
  645. else:
  646. self.log.trace(f"Removed message {message.item_id} after Matrix redaction")
  647. return
  648. raise NotImplementedError("No message or reaction found for redaction")
  649. async def handle_matrix_typing(self, users: set[UserID]) -> None:
  650. if users == self._typing:
  651. return
  652. old_typing = self._typing
  653. self._typing = users
  654. await self._handle_matrix_typing(old_typing - users, TypingStatus.OFF)
  655. await self._handle_matrix_typing(users - old_typing, TypingStatus.TEXT)
  656. async def _handle_matrix_typing(self, users: set[UserID], status: TypingStatus) -> None:
  657. for mxid in users:
  658. user = await u.User.get_by_mxid(mxid, create=False)
  659. if (
  660. not user
  661. or not await user.is_logged_in()
  662. or user.remote_typing_status == status
  663. or not user.is_connected
  664. ):
  665. continue
  666. user.remote_typing_status = None
  667. await user.mqtt.indicate_activity(self.thread_id, status)
  668. async def handle_matrix_leave(self, user: u.User) -> None:
  669. if not await user.is_logged_in():
  670. return
  671. if self.is_direct:
  672. self.log.info(f"{user.mxid} left private chat portal with {self.other_user_pk}")
  673. if user.igpk == self.receiver:
  674. self.log.info(
  675. f"{user.mxid} was the recipient of this portal. Cleaning up and deleting..."
  676. )
  677. await self.cleanup_and_delete()
  678. else:
  679. self.log.debug(f"{user.mxid} left portal to {self.thread_id}")
  680. # TODO cleanup if empty
  681. # endregion
  682. # region Instagram event handling
  683. async def _reupload_instagram_media(
  684. self, source: u.User, media: RegularMediaItem, intent: IntentAPI
  685. ) -> MediaMessageEventContent:
  686. if media.media_type == MediaType.IMAGE:
  687. image = media.best_image
  688. if not image:
  689. raise ValueError("Attachment not available: didn't find photo URL")
  690. url = image.url
  691. msgtype = MessageType.IMAGE
  692. info = ImageInfo(height=image.height, width=image.width)
  693. elif media.media_type == MediaType.VIDEO:
  694. video = media.best_video
  695. if not video:
  696. raise ValueError("Attachment not available: didn't find video URL")
  697. url = video.url
  698. msgtype = MessageType.VIDEO
  699. info = VideoInfo(height=video.height, width=video.width)
  700. elif media.media_type == MediaType.CAROUSEL:
  701. raise ValueError(
  702. "Carousel media is not currently supported, "
  703. "please view the post on Instagram via the link below"
  704. )
  705. else:
  706. raise ValueError(
  707. f"Attachment not available: unsupported media type {media.media_type.human_name}"
  708. )
  709. return await self._reupload_instagram_file(source, url, msgtype, info, intent)
  710. async def _reupload_instagram_animated(
  711. self, source: u.User, media: AnimatedMediaItem, intent: IntentAPI
  712. ) -> MediaMessageEventContent:
  713. url = media.images.fixed_height.webp
  714. info = ImageInfo(
  715. height=int(media.images.fixed_height.height),
  716. width=int(media.images.fixed_height.width),
  717. )
  718. return await self._reupload_instagram_file(source, url, MessageType.IMAGE, info, intent)
  719. async def _reupload_instagram_xma(
  720. self, source: u.User, media: XMAMediaShareItem, intent: IntentAPI
  721. ) -> MediaMessageEventContent:
  722. url = media.preview_url
  723. info = ImageInfo(
  724. mimetype=media.preview_url_mime_type,
  725. )
  726. return await self._reupload_instagram_file(source, url, MessageType.IMAGE, info, intent)
  727. async def _reupload_instagram_voice(
  728. self, source: u.User, media: VoiceMediaItem, intent: IntentAPI
  729. ) -> MediaMessageEventContent:
  730. async def convert_to_ogg(data, mimetype):
  731. converted = await ffmpeg.convert_bytes(
  732. data, ".ogg", output_args=("-c:a", "libopus"), input_mime=mimetype
  733. )
  734. return converted, "audio/ogg"
  735. url = media.media.audio.audio_src
  736. info = AudioInfo(duration=media.media.audio.duration)
  737. waveform = [int(p * 1024) for p in media.media.audio.waveform_data]
  738. content = await self._reupload_instagram_file(
  739. source, url, MessageType.AUDIO, info, intent, convert_to_ogg
  740. )
  741. content["org.matrix.msc1767.audio"] = {
  742. "duration": media.media.audio.duration,
  743. "waveform": waveform,
  744. }
  745. content["org.matrix.msc3245.voice"] = {}
  746. return content
  747. async def _download_instagram_file(
  748. self, source: u.User, url: str
  749. ) -> tuple[Optional[bytes], str]:
  750. async with source.client.raw_http_get(url) as resp:
  751. try:
  752. length = int(resp.headers["Content-Length"])
  753. except KeyError:
  754. # TODO can the download be short-circuited if there's too much data?
  755. self.log.warning(
  756. "Got file download response with no Content-Length header,"
  757. "reading data dangerously"
  758. )
  759. length = 0
  760. if length > self.matrix.media_config.upload_size:
  761. self.log.debug(
  762. f"{url} was too large ({length} > {self.matrix.media_config.upload_size})"
  763. )
  764. raise ValueError("Attachment not available: too large")
  765. data = await resp.read()
  766. if not data:
  767. return None, ""
  768. mimetype = resp.headers["Content-Type"] or magic.from_buffer(data, mime=True)
  769. return data, mimetype
  770. async def _reupload_instagram_file(
  771. self,
  772. source: u.User,
  773. url: str,
  774. msgtype: MessageType | None,
  775. info: ImageInfo | VideoInfo | AudioInfo,
  776. intent: IntentAPI,
  777. convert_fn: Callable[[bytes, str], Awaitable[tuple[bytes, str]]] | None = None,
  778. allow_encrypt: bool = True,
  779. ) -> MediaMessageEventContent:
  780. data, mimetype = await self._download_instagram_file(source, url)
  781. assert data is not None
  782. info.mimetype = mimetype
  783. # Run the conversion function on the data.
  784. if convert_fn is not None:
  785. data, info.mimetype = await convert_fn(data, info.mimetype)
  786. if info.mimetype.startswith("image/") and not info.width and not info.height:
  787. with BytesIO(data) as inp, Image.open(inp) as img:
  788. info.width, info.height = img.size
  789. info.size = len(data)
  790. extension = {
  791. "image/webp": ".webp",
  792. "image/jpeg": ".jpg",
  793. "video/mp4": ".mp4",
  794. "audio/mp4": ".m4a",
  795. "audio/ogg": ".ogg",
  796. }.get(info.mimetype)
  797. extension = extension or mimetypes.guess_extension(info.mimetype) or ""
  798. file_name = f"{msgtype.value[2:]}{extension}" if msgtype else None
  799. upload_mime_type = info.mimetype
  800. upload_file_name = file_name
  801. decryption_info = None
  802. if allow_encrypt and self.encrypted and encrypt_attachment:
  803. data, decryption_info = encrypt_attachment(data)
  804. upload_mime_type = "application/octet-stream"
  805. upload_file_name = None
  806. mxc = await intent.upload_media(
  807. data,
  808. mime_type=upload_mime_type,
  809. filename=upload_file_name,
  810. async_upload=self.config["homeserver.async_media"],
  811. )
  812. if decryption_info:
  813. decryption_info.url = mxc
  814. mxc = None
  815. return MediaMessageEventContent(
  816. body=file_name,
  817. external_url=url,
  818. url=mxc,
  819. file=decryption_info,
  820. info=info,
  821. msgtype=msgtype,
  822. )
  823. def _get_instagram_media_info(self, item: ThreadItem) -> tuple[MediaUploadFunc, MediaData]:
  824. # TODO maybe use a dict and item.item_type instead of a ton of ifs
  825. method = self._reupload_instagram_media
  826. if (
  827. item.xma_media_share
  828. or item.xma_story_share
  829. or item.xma_reel_share
  830. or item.xma_reel_mention
  831. or item.generic_xma
  832. ):
  833. media_data = (
  834. item.xma_media_share
  835. or item.xma_story_share
  836. or item.xma_reel_share
  837. or item.xma_reel_mention
  838. or item.generic_xma
  839. )[0]
  840. method = self._reupload_instagram_xma
  841. elif item.media:
  842. media_data = item.media
  843. elif item.visual_media:
  844. media_data = item.visual_media.media
  845. elif item.animated_media:
  846. media_data = item.animated_media
  847. method = self._reupload_instagram_animated
  848. elif item.voice_media:
  849. media_data = item.voice_media
  850. method = self._reupload_instagram_voice
  851. elif item.reel_share:
  852. media_data = item.reel_share.media
  853. elif item.story_share:
  854. media_data = item.story_share.media
  855. elif item.clip:
  856. media_data = item.clip.clip
  857. elif item.felix_share and item.felix_share.video:
  858. media_data = item.felix_share.video
  859. elif item.media_share:
  860. media_data = item.media_share
  861. elif item.direct_media_share:
  862. media_data = item.direct_media_share.media
  863. else:
  864. self.log.debug(f"Unknown media type in {item}")
  865. raise ValueError("Attachment not available: unsupported media type")
  866. if not media_data:
  867. self.log.debug(f"Didn't get media_data in {item}")
  868. raise ValueError("Attachment not available: unsupported media type")
  869. elif isinstance(media_data, ExpiredMediaItem):
  870. self.log.debug(f"Expired media in item {item}")
  871. raise ValueError("Attachment not available: media expired")
  872. return method, media_data
  873. async def _convert_instagram_media(
  874. self, source: u.User, intent: IntentAPI, item: ThreadItem
  875. ) -> ConvertedMessage:
  876. try:
  877. reupload_func, media_data = self._get_instagram_media_info(item)
  878. content = await reupload_func(source, media_data, intent)
  879. except ValueError as e:
  880. content = TextMessageEventContent(body=str(e), msgtype=MessageType.NOTICE)
  881. except Exception:
  882. self.log.warning("Failed to upload media", exc_info=True)
  883. content = TextMessageEventContent(
  884. body="Attachment not available: failed to copy file", msgtype=MessageType.NOTICE
  885. )
  886. await self._add_instagram_reply(content, item.replied_to_message)
  887. return EventType.ROOM_MESSAGE, content
  888. # TODO this might be unused
  889. async def _convert_instagram_media_share(
  890. self, source: u.User, intent: IntentAPI, item: ThreadItem
  891. ) -> list[ConvertedMessage]:
  892. item_type_name = None
  893. if item.media_share:
  894. share_item = item.media_share
  895. elif item.clip:
  896. share_item = item.clip.clip
  897. item_type_name = "clip"
  898. elif item.felix_share and item.felix_share.video:
  899. share_item = item.felix_share.video
  900. elif item.story_share:
  901. share_item = item.story_share.media
  902. item_type_name = "story"
  903. elif item.direct_media_share:
  904. share_item = item.direct_media_share.media
  905. else:
  906. self.log.debug("No media share to bridge")
  907. return []
  908. item_type_name = item_type_name or share_item.media_type.human_name
  909. user_text = f"@{share_item.user.username}"
  910. user_link = (
  911. f'<a href="https://www.instagram.com/{share_item.user.username}/">{user_text}</a>'
  912. )
  913. prefix = TextMessageEventContent(
  914. msgtype=MessageType.NOTICE,
  915. format=Format.HTML,
  916. body=f"Sent {user_text}'s {item_type_name}",
  917. formatted_body=f"Sent {user_link}'s {item_type_name}",
  918. )
  919. if item.direct_media_share and item.direct_media_share.media_share_type == "tag":
  920. tagged_user_id = item.direct_media_share.tagged_user_id
  921. if tagged_user_id == source.igpk and share_item.user.pk == self.other_user_pk:
  922. prefix.body = prefix.formatted_body = "Tagged you in their post"
  923. elif share_item.user.pk == source.igpk and tagged_user_id == self.other_user_pk:
  924. prefix.body = prefix.formatted_body = "Tagged them in your post"
  925. _, content = await self._convert_instagram_media(source, intent, item)
  926. external_url = f"https://www.instagram.com/p/{share_item.code}/"
  927. if share_item.caption:
  928. caption_body = (
  929. f"> {share_item.caption.user.username}: {share_item.caption.text}\n\n"
  930. f"{external_url}"
  931. )
  932. caption_formatted_body = (
  933. f"<blockquote><strong>{share_item.caption.user.username}</strong>"
  934. f" {html.escape(share_item.caption.text)}</blockquote>"
  935. f'<a href="{external_url}">instagram.com/p/{share_item.code}</a>'
  936. )
  937. else:
  938. caption_body = external_url
  939. caption_formatted_body = (
  940. f'<a href="{external_url}">instagram.com/p/{share_item.code}</a>'
  941. )
  942. caption = TextMessageEventContent(
  943. msgtype=MessageType.TEXT,
  944. body=caption_body,
  945. formatted_body=caption_formatted_body,
  946. format=Format.HTML,
  947. external_url=external_url,
  948. )
  949. if self.bridge.config["bridge.caption_in_message"]:
  950. if isinstance(content, TextMessageEventContent):
  951. content.ensure_has_html()
  952. prefix.ensure_has_html()
  953. caption.ensure_has_html()
  954. combined = TextMessageEventContent(
  955. msgtype=MessageType.TEXT,
  956. body="\n".join((content.body, prefix.body, caption.body)),
  957. formatted_body=(
  958. f"<p><b>{content.formatted_body}</b></p>"
  959. f"<p><i>{prefix.formatted_body}</p>"
  960. f"<p>{caption.formatted_body}</p>"
  961. ),
  962. format=Format.HTML,
  963. external_url=external_url,
  964. )
  965. else:
  966. prefix.ensure_has_html()
  967. caption.ensure_has_html()
  968. combined_body = "\n".join((prefix.body, caption.body))
  969. combined_formatted_body = (
  970. f"<p><i>{prefix.formatted_body}</i></p><p>{caption.formatted_body}</p>"
  971. )
  972. combined = content
  973. combined["filename"] = content.body
  974. combined.body = combined_body
  975. combined["format"] = str(Format.HTML)
  976. combined["org.matrix.msc1767.caption"] = {
  977. "org.matrix.msc1767.text": combined_body,
  978. "org.matrix.msc1767.html": combined_formatted_body,
  979. }
  980. combined["formatted_body"] = combined_formatted_body
  981. return [(EventType.ROOM_MESSAGE, combined)]
  982. else:
  983. return [
  984. (EventType.ROOM_MESSAGE, prefix),
  985. (EventType.ROOM_MESSAGE, content),
  986. (EventType.ROOM_MESSAGE, caption),
  987. ]
  988. async def _convert_instagram_xma_media_share(
  989. self, source: u.User, intent: IntentAPI, item: ThreadItem
  990. ) -> list[ConvertedMessage]:
  991. # N.B. _get_instagram_media_info also only supports downloading the first xma item
  992. xma_list = (
  993. item.xma_media_share
  994. or item.xma_story_share
  995. or item.xma_reel_share
  996. or item.xma_reel_mention
  997. or item.generic_xma
  998. )
  999. media = xma_list[0]
  1000. if len(xma_list) != 1:
  1001. self.log.warning(f"Item {item.item_id} has multiple xma media share parts")
  1002. if media.xma_layout_type not in (0, 4):
  1003. self.log.warning(f"Unrecognized xma layout type {media.xma_layout_type}")
  1004. _, content = await self._convert_instagram_media(source, intent, item)
  1005. # Post shares (layout type 0): media title text
  1006. # Reel shares/replies/reactions (layout type 4): item text
  1007. caption_text = media.title_text or item.text or ""
  1008. if media.subtitle_text:
  1009. caption_text = (
  1010. f"{caption_text}\n{media.subtitle_text}" if caption_text else media.subtitle_text
  1011. )
  1012. if media.target_url:
  1013. caption_body = (
  1014. f"> {caption_text}\n\n{media.target_url}" if caption_text else media.target_url
  1015. )
  1016. else:
  1017. caption_body = f"> {caption_text}"
  1018. escaped_caption_text = html.escape(caption_text).replace("\n", "<br>")
  1019. escaped_header_text = html.escape(media.header_title_text or "")
  1020. # For post shares, the media title starts with the username, which is also the header.
  1021. # That part should be bolded.
  1022. if (
  1023. escaped_header_text
  1024. and escaped_caption_text
  1025. and escaped_caption_text.startswith(escaped_header_text)
  1026. ):
  1027. escaped_caption_text = (
  1028. f"<strong>{escaped_header_text}</strong>"
  1029. f"{escaped_caption_text[len(escaped_header_text):]}"
  1030. )
  1031. if item.message_item_type == "animated_media":
  1032. anim = await self._reupload_instagram_file(
  1033. source,
  1034. url=item.animated_media.images.fixed_height.webp,
  1035. msgtype=MessageType.IMAGE,
  1036. info=ImageInfo(
  1037. width=int(item.animated_media.images.fixed_height.width),
  1038. height=int(item.animated_media.images.fixed_height.height),
  1039. ),
  1040. intent=intent,
  1041. )
  1042. inline_img = (
  1043. f'<img src="{anim.url}" width={anim.info.width} height={anim.info.height}/>'
  1044. )
  1045. escaped_caption_text = (
  1046. f"{escaped_caption_text}<br/>{inline_img}" if escaped_caption_text else inline_img
  1047. )
  1048. caption_formatted_body = (
  1049. f"<blockquote>{escaped_caption_text}</blockquote>" if escaped_caption_text else ""
  1050. )
  1051. if media.target_url:
  1052. target_url_pretty = str(URL(media.target_url).with_query(None)).replace(
  1053. "https://www.", ""
  1054. )
  1055. caption_formatted_body += (
  1056. f'<p><a href="{media.target_url}">{target_url_pretty}</a></p>'
  1057. )
  1058. # Add auxiliary text as prefix for caption
  1059. if item.auxiliary_text:
  1060. caption_formatted_body = (
  1061. f"<p>{html.escape(item.auxiliary_text)}</p>{caption_formatted_body}"
  1062. )
  1063. caption_body = f"{item.auxiliary_text}\n\n{caption_body}"
  1064. caption = TextMessageEventContent(
  1065. msgtype=MessageType.TEXT,
  1066. body=caption_body,
  1067. formatted_body=caption_formatted_body,
  1068. format=Format.HTML,
  1069. )
  1070. if media.target_url:
  1071. content.external_url = media.target_url
  1072. caption.external_url = media.target_url
  1073. if self.bridge.config["bridge.caption_in_message"]:
  1074. if isinstance(content, TextMessageEventContent):
  1075. content.ensure_has_html()
  1076. caption.ensure_has_html()
  1077. content.body += f"\n\n{caption.body}"
  1078. content.formatted_body = (
  1079. f"<p><b>{content.formatted_body}</b></p>{caption.formatted_body}"
  1080. )
  1081. else:
  1082. content["filename"] = content.body
  1083. content.body = caption.body
  1084. content["format"] = str(Format.HTML)
  1085. content["formatted_body"] = caption.formatted_body
  1086. content["org.matrix.msc1767.caption"] = {
  1087. "org.matrix.msc1767.text": content.body,
  1088. "org.matrix.msc1767.html": content["formatted_body"],
  1089. }
  1090. return [(EventType.ROOM_MESSAGE, content)]
  1091. else:
  1092. return [(EventType.ROOM_MESSAGE, content), (EventType.ROOM_MESSAGE, caption)]
  1093. # TODO this is probably unused
  1094. async def _convert_instagram_reel_share(
  1095. self, source: u.User, intent: IntentAPI, item: ThreadItem
  1096. ) -> list[ConvertedMessage]:
  1097. assert item.reel_share
  1098. media = item.reel_share.media
  1099. prefix_html = None
  1100. if item.reel_share.type == ReelShareType.REPLY:
  1101. if item.reel_share.reel_owner_id == source.igpk:
  1102. prefix = "Replied to your story"
  1103. else:
  1104. username = media.user.username
  1105. prefix = f"Sent @{username}'s story"
  1106. user_link = f'<a href="https://www.instagram.com/{username}/">@{username}</a>'
  1107. prefix_html = f"Sent {user_link}'s story"
  1108. elif item.reel_share.type == ReelShareType.REACTION:
  1109. if item.reel_share.reel_owner_id == source.igpk:
  1110. prefix = "Reacted to your story"
  1111. elif item.user_id == source.igpk:
  1112. prefix = "You reacted to their story"
  1113. else:
  1114. prefix = "Reacted to a story"
  1115. elif item.reel_share.type == ReelShareType.MENTION:
  1116. if item.reel_share.mentioned_user_id == source.igpk:
  1117. prefix = "Mentioned you in their story"
  1118. else:
  1119. prefix = "You mentioned them in your story"
  1120. else:
  1121. self.log.debug(f"Unsupported reel share type {item.reel_share.type}")
  1122. return []
  1123. prefix_content = TextMessageEventContent(msgtype=MessageType.NOTICE, body=prefix)
  1124. if prefix_html:
  1125. prefix_content.format = Format.HTML
  1126. prefix_content.formatted_body = prefix_html
  1127. caption_content = TextMessageEventContent(
  1128. msgtype=MessageType.TEXT, body=item.reel_share.text
  1129. )
  1130. if not caption_content.body and isinstance(media, MediaShareItem):
  1131. caption_content.body = media.caption.text if media.caption else ""
  1132. if not caption_content.body:
  1133. caption_content.body = "<no caption>"
  1134. media_content = None
  1135. fake_item_id = f"fi.mau.instagram.reel_share.{item.user_id}.{media.pk}"
  1136. if isinstance(media, ExpiredMediaItem):
  1137. media_content = TextMessageEventContent(
  1138. msgtype=MessageType.NOTICE, body="Story expired"
  1139. )
  1140. else:
  1141. existing = await DBMessage.get_by_item_id(fake_item_id, self.receiver)
  1142. if existing:
  1143. # If the user already reacted or replied to the same reel share item,
  1144. # use a Matrix reply instead of reposting the image.
  1145. caption_content.set_reply(existing.mxid)
  1146. else:
  1147. _, media_content = await self._convert_instagram_media(source, intent, item)
  1148. if self.bridge.config["bridge.caption_in_message"]:
  1149. if media_content:
  1150. if isinstance(media_content, TextMessageEventContent):
  1151. media_content.ensure_has_html()
  1152. prefix_content.ensure_has_html()
  1153. caption_content.ensure_has_html()
  1154. combined = TextMessageEventContent(
  1155. msgtype=MessageType.TEXT,
  1156. body="\n".join(
  1157. (media_content.body, prefix_content.body, caption_content.body)
  1158. ),
  1159. formatted_body=(
  1160. f"<p><b>{media_content.formatted_body}</b></p>"
  1161. f"<p><i>{prefix_content.formatted_body}</i></p>"
  1162. f"<p>{caption_content.formatted_body}</p>"
  1163. ),
  1164. format=Format.HTML,
  1165. )
  1166. else:
  1167. prefix_content.ensure_has_html()
  1168. caption_content.ensure_has_html()
  1169. combined_body = "\n".join((prefix_content.body, caption_content.body))
  1170. combined_formatted_body = (
  1171. f"<p><i>{prefix_content.formatted_body}</i></p>"
  1172. f"<p>{caption_content.formatted_body}</p>"
  1173. )
  1174. combined = media_content
  1175. combined["filename"] = combined.body
  1176. combined.body = combined_body
  1177. combined["format"] = str(Format.HTML)
  1178. combined["org.matrix.msc1767.caption"] = {
  1179. "org.matrix.msc1767.text": combined_body,
  1180. "org.matrix.msc1767.html": combined_formatted_body,
  1181. }
  1182. combined["formatted_body"] = combined_formatted_body
  1183. else:
  1184. combined = caption_content
  1185. return [(EventType.ROOM_MESSAGE, combined)]
  1186. else:
  1187. await self._send_message(intent, prefix_content, timestamp=item.timestamp_ms)
  1188. converted: list[ConvertedMessage] = []
  1189. if media_content:
  1190. converted.append((EventType.ROOM_MESSAGE, media_content))
  1191. converted.append((EventType.ROOM_MESSAGE, caption_content))
  1192. return converted
  1193. async def _convert_instagram_link(
  1194. self,
  1195. source: u.User,
  1196. intent: IntentAPI,
  1197. item: ThreadItem,
  1198. ) -> ConvertedMessage:
  1199. content = TextMessageEventContent(msgtype=MessageType.TEXT, body=item.link.text)
  1200. link = item.link.link_context
  1201. preview = {
  1202. "og:url": link.link_url,
  1203. "og:title": link.link_title,
  1204. "og:description": link.link_summary,
  1205. }
  1206. if link.link_image_url:
  1207. reuploaded = await self._reupload_instagram_file(
  1208. source, link.link_image_url, msgtype=None, info=ImageInfo(), intent=intent
  1209. )
  1210. preview["og:image"] = reuploaded.url
  1211. preview["og:image:type"] = reuploaded.info.mimetype
  1212. preview["og:image:width"] = reuploaded.info.width
  1213. preview["og:image:height"] = reuploaded.info.height
  1214. preview["matrix:image:size"] = reuploaded.info.size
  1215. if reuploaded.file:
  1216. preview["beeper:image:encryption"] = reuploaded.file.serialize()
  1217. preview = {k: v for k, v in preview.items() if v}
  1218. content["com.beeper.linkpreviews"] = [preview] if "og:title" in preview else []
  1219. await self._add_instagram_reply(content, item.replied_to_message)
  1220. return EventType.ROOM_MESSAGE, content
  1221. async def _convert_expired_placeholder(
  1222. self, source: u.User, item: ThreadItem, action: str
  1223. ) -> ConvertedMessage:
  1224. if item.user_id == source.igpk:
  1225. prefix = f"{action} your story"
  1226. elif item.user_id == source.igpk:
  1227. prefix = f"You {action.lower()} their story"
  1228. else:
  1229. prefix = f"{action} a story"
  1230. body = f"{prefix}\n\nNo longer available"
  1231. html = f"<p>{prefix}</p><p><i>No longer available</i></p>"
  1232. content = TextMessageEventContent(
  1233. msgtype=MessageType.NOTICE, body=body, format=Format.HTML, formatted_body=html
  1234. )
  1235. return EventType.ROOM_MESSAGE, content
  1236. async def _convert_instagram_text(self, item: ThreadItem, text: str) -> ConvertedMessage:
  1237. content = TextMessageEventContent(msgtype=MessageType.TEXT, body=text)
  1238. content["com.beeper.linkpreviews"] = []
  1239. await self._add_instagram_reply(content, item.replied_to_message)
  1240. return EventType.ROOM_MESSAGE, content
  1241. async def _convert_instagram_unhandled(self, item: ThreadItem) -> ConvertedMessage:
  1242. content = TextMessageEventContent(
  1243. msgtype=MessageType.NOTICE, body=f"Unsupported message type {item.item_type.value}"
  1244. )
  1245. await self._add_instagram_reply(content, item.replied_to_message)
  1246. return EventType.ROOM_MESSAGE, content
  1247. async def _convert_instagram_location(self, item: ThreadItem) -> ConvertedMessage | None:
  1248. loc = item.location
  1249. if not loc or not loc.lng or not loc.lat:
  1250. # TODO handle somehow
  1251. return None
  1252. long_char = "E" if loc.lng > 0 else "W"
  1253. lat_char = "N" if loc.lat > 0 else "S"
  1254. body = (
  1255. f"{loc.name} - {round(abs(loc.lat), 4)}° {lat_char}, "
  1256. f"{round(abs(loc.lng), 4)}° {long_char}"
  1257. )
  1258. url = f"https://www.openstreetmap.org/#map=15/{loc.lat}/{loc.lng}"
  1259. external_url = None
  1260. if loc.external_source == "facebook_places":
  1261. external_url = f"https://www.facebook.com/{loc.short_name}-{loc.facebook_places_id}"
  1262. content = LocationMessageEventContent(
  1263. msgtype=MessageType.LOCATION,
  1264. geo_uri=f"geo:{loc.lat},{loc.lng}",
  1265. body=f"Location: {body}\n{url}",
  1266. external_url=external_url,
  1267. )
  1268. content["format"] = str(Format.HTML)
  1269. content["formatted_body"] = f"Location: <a href='{url}'>{body}</a>"
  1270. await self._add_instagram_reply(content, item.replied_to_message)
  1271. return EventType.ROOM_MESSAGE, content
  1272. async def _convert_instagram_profile(self, item: ThreadItem) -> ConvertedMessage:
  1273. username = item.profile.username
  1274. user_link = f'<a href="https://www.instagram.com/{username}/">@{username}</a>'
  1275. text = f"Shared @{username}'s profile"
  1276. html = f"Shared {user_link}'s profile"
  1277. content = TextMessageEventContent(
  1278. msgtype=MessageType.TEXT, format=Format.HTML, body=text, formatted_body=html
  1279. )
  1280. await self._add_instagram_reply(content, item.replied_to_message)
  1281. return EventType.ROOM_MESSAGE, content
  1282. async def _convert_instagram_xma_profile_share(
  1283. self, item: ThreadItem
  1284. ) -> list[ConvertedMessage]:
  1285. assert item.xma_profile
  1286. profile_messages = []
  1287. for profile in item.xma_profile:
  1288. username = profile.header_title_text
  1289. user_link = f'<a href="{profile.target_url}">@{username}</a>'
  1290. text = f"Shared @{username}'s profile"
  1291. html = f"Shared {user_link}'s profile"
  1292. content = TextMessageEventContent(
  1293. msgtype=MessageType.TEXT, format=Format.HTML, body=text, formatted_body=html
  1294. )
  1295. await self._add_instagram_reply(content, item.replied_to_message)
  1296. profile_messages.append((EventType.ROOM_MESSAGE, content))
  1297. return profile_messages
  1298. async def _add_instagram_reply(
  1299. self, content: MessageEventContent, reply_to: ThreadItem | None
  1300. ) -> None:
  1301. if not reply_to:
  1302. return
  1303. message = await DBMessage.get_by_item_id(reply_to.item_id, self.receiver)
  1304. if not message:
  1305. return
  1306. content.set_reply(message.mxid)
  1307. if not isinstance(content, TextMessageEventContent):
  1308. return
  1309. try:
  1310. evt = await self.main_intent.get_event(message.mx_room, message.mxid)
  1311. except (MNotFound, MForbidden):
  1312. evt = None
  1313. if not evt:
  1314. return
  1315. if evt.type == EventType.ROOM_ENCRYPTED:
  1316. try:
  1317. evt = await self.matrix.e2ee.decrypt(evt, wait_session_timeout=0)
  1318. except SessionNotFound:
  1319. return
  1320. if isinstance(evt.content, TextMessageEventContent):
  1321. evt.content.trim_reply_fallback()
  1322. content.set_reply(evt)
  1323. async def handle_instagram_item(
  1324. self, source: u.User, sender: p.Puppet, item: MessageSyncMessage
  1325. ):
  1326. client_context = item.client_context
  1327. link_client_context = item.link.client_context if item.link else None
  1328. cc = client_context
  1329. if link_client_context:
  1330. if not client_context:
  1331. cc = f"link:{link_client_context}"
  1332. elif client_context != link_client_context:
  1333. cc = f"{client_context}/link:{link_client_context}"
  1334. if client_context and client_context in self._reqid_dedup:
  1335. self.log.debug(
  1336. f"Ignoring message {item.item_id} ({cc}) by {item.user_id}"
  1337. " as it was sent by us (client_context in dedup queue)"
  1338. )
  1339. return []
  1340. elif link_client_context and link_client_context in self._reqid_dedup:
  1341. self.log.debug(
  1342. f"Ignoring message {item.item_id} ({cc}) by {item.user_id}"
  1343. " as it was sent by us (link.client_context in dedup queue)"
  1344. )
  1345. return []
  1346. # Check in-memory queues for duplicates
  1347. if item.item_id in self._msgid_dedup:
  1348. self.log.debug(
  1349. f"Ignoring message {item.item_id} ({item.client_context}) by {item.user_id}"
  1350. " as it was already handled (message.id in dedup queue)"
  1351. )
  1352. return
  1353. self._msgid_dedup.appendleft(item.item_id)
  1354. # Check database for duplicates
  1355. if await DBMessage.get_by_item_id(item.item_id, self.receiver) is not None:
  1356. self.log.debug(
  1357. f"Ignoring message {item.item_id} ({item.client_context}) by {item.user_id}"
  1358. " as it was already handled (message.id in database)"
  1359. )
  1360. return
  1361. self.log.debug(
  1362. f"Handling Instagram message {item.item_id} ({item.client_context}) by {item.user_id}"
  1363. )
  1364. if not self.mxid:
  1365. thread = await source.client.get_thread(item.thread_id)
  1366. mxid = await self.create_matrix_room(source, thread.thread)
  1367. if not mxid:
  1368. # Failed to create
  1369. return
  1370. if self.config["bridge.backfill.enable"]:
  1371. if self.config["bridge.backfill.msc2716"]:
  1372. await self.enqueue_immediate_backfill(source, 0)
  1373. intent = sender.intent_for(self)
  1374. asyncio.create_task(intent.set_typing(self.mxid, is_typing=False))
  1375. event_ids = []
  1376. for event_type, content in await self.convert_instagram_item(source, sender, item):
  1377. event_ids.append(
  1378. await self._send_message(
  1379. intent, content, event_type=event_type, timestamp=item.timestamp_ms
  1380. )
  1381. )
  1382. event_ids = [event_id for event_id in event_ids if event_id]
  1383. if not event_ids:
  1384. self.log.warning(f"Unhandled Instagram message {item.item_id}")
  1385. return
  1386. self.log.debug(f"Handled Instagram message {item.item_id} -> {event_ids}")
  1387. await DBMessage(
  1388. mxid=event_ids[-1],
  1389. mx_room=self.mxid,
  1390. item_id=item.item_id,
  1391. client_context=item.client_context,
  1392. receiver=self.receiver,
  1393. sender=sender.igpk,
  1394. ig_timestamp=item.timestamp,
  1395. ).insert()
  1396. await self._send_delivery_receipt(event_ids[-1])
  1397. async def convert_instagram_item(
  1398. self, source: u.User, sender: p.Puppet, item: ThreadItem
  1399. ) -> list[ConvertedMessage]:
  1400. if not isinstance(item, ThreadItem):
  1401. # Parsing these items failed, they should have been logged already
  1402. return []
  1403. try:
  1404. return await self._convert_instagram_item(source, sender, item)
  1405. except Exception:
  1406. self.log.exception("Fatal error converting Instagram item")
  1407. self.log.trace("Item content: %s", item.serialize())
  1408. return []
  1409. async def _convert_instagram_item(
  1410. self, source: u.User, sender: p.Puppet, item: ThreadItem
  1411. ) -> list[ConvertedMessage]:
  1412. intent = sender.intent_for(self)
  1413. if (
  1414. item.xma_media_share
  1415. or item.xma_reel_share
  1416. or item.xma_reel_mention
  1417. or item.xma_story_share
  1418. or item.generic_xma
  1419. ):
  1420. return await self._convert_instagram_xma_media_share(source, intent, item)
  1421. converted: list[ConvertedMessage] = []
  1422. handle_text = True
  1423. if item.media or item.animated_media or item.voice_media or item.visual_media:
  1424. converted.append(await self._convert_instagram_media(source, intent, item))
  1425. elif item.location:
  1426. if loc_content := await self._convert_instagram_location(item):
  1427. converted.append(loc_content)
  1428. elif item.profile:
  1429. converted.append(await self._convert_instagram_profile(item))
  1430. elif item.xma_profile:
  1431. converted.extend(await self._convert_instagram_xma_profile_share(item))
  1432. elif item.reel_share:
  1433. converted.extend(await self._convert_instagram_reel_share(source, intent, item))
  1434. elif (
  1435. item.media_share
  1436. or item.direct_media_share
  1437. or item.story_share
  1438. or item.clip
  1439. or item.felix_share
  1440. ):
  1441. converted.extend(await self._convert_instagram_media_share(source, intent, item))
  1442. elif item.item_type == ThreadItemType.EXPIRED_PLACEHOLDER:
  1443. if item.message_item_type == "reaction":
  1444. action = "Reacted to"
  1445. else:
  1446. action = "Shared"
  1447. msg_type, expired = await self._convert_expired_placeholder(source, item, action)
  1448. if self.bridge.config["bridge.caption_in_message"] and item.text:
  1449. _, text = await self._convert_instagram_text(item, item.text)
  1450. expired.ensure_has_html()
  1451. text.ensure_has_html()
  1452. combined = TextMessageEventContent(
  1453. msgtype=MessageType.TEXT,
  1454. body="\n".join((expired.body, text.body)),
  1455. formatted_body=f"{expired.formatted_body}<p>{text.formatted_body}</p>",
  1456. format=Format.HTML,
  1457. )
  1458. handle_text = False
  1459. converted.append((msg_type, combined))
  1460. else:
  1461. converted.append((msg_type, expired))
  1462. elif item.action_log:
  1463. # These probably don't need to be bridged
  1464. self.log.debug(f"Ignoring action log message {item.item_id}")
  1465. return []
  1466. # TODO handle item.clip?
  1467. # TODO should these be put into a caption?
  1468. if handle_text and item.text:
  1469. converted.append(await self._convert_instagram_text(item, item.text))
  1470. elif item.like:
  1471. # We handle likes as text because Matrix clients do big emoji on their own.
  1472. converted.append(await self._convert_instagram_text(item, item.like))
  1473. elif item.link:
  1474. converted.append(await self._convert_instagram_link(source, intent, item))
  1475. if len(converted) == 0:
  1476. self.log.debug(f"Unhandled Instagram message {item.item_id}")
  1477. converted.append(await self._convert_instagram_unhandled(item))
  1478. return converted
  1479. def _deterministic_event_id(
  1480. self, sender: p.Puppet, item_id: str, part_name: int | None = None
  1481. ) -> EventID:
  1482. hash_content = f"{self.mxid}/instagram/{sender.igpk}/{item_id}"
  1483. if part_name:
  1484. hash_content += f"/{part_name}"
  1485. hashed = hashlib.sha256(hash_content.encode("utf-8")).digest()
  1486. b64hash = base64.urlsafe_b64encode(hashed).decode("utf-8").rstrip("=")
  1487. return EventID(f"${b64hash}:telegram.org")
  1488. async def handle_instagram_remove(self, item_id: str) -> None:
  1489. message = await DBMessage.get_by_item_id(item_id, self.receiver)
  1490. if message is None:
  1491. return
  1492. await message.delete()
  1493. if message.mxid:
  1494. sender = await p.Puppet.get_by_pk(message.sender)
  1495. try:
  1496. await sender.intent_for(self).redact(self.mxid, message.mxid)
  1497. except MForbidden:
  1498. await self.main_intent.redact(self.mxid, message.mxid)
  1499. self.log.debug(f"Redacted {message.mxid} after Instagram unsend")
  1500. async def handle_instagram_reaction(self, item: ThreadItem, remove: bool) -> None:
  1501. sender = await p.Puppet.get_by_pk(item.new_reaction.sender_id)
  1502. message = await DBMessage.get_by_item_id(item.item_id, self.receiver)
  1503. if not message:
  1504. self.log.debug(f"Dropping reaction by {sender.pk} to unknown message {item.item_id}")
  1505. return
  1506. emoji = item.new_reaction.emoji
  1507. async with self._reaction_lock:
  1508. existing = await DBReaction.get_by_item_id(item.item_id, self.receiver, sender.pk)
  1509. if not existing and remove:
  1510. self.log.debug(
  1511. f"Ignoring duplicate reaction removal by {sender.pk} to {item.item_id}"
  1512. )
  1513. return
  1514. elif not remove and existing and existing.reaction == emoji:
  1515. self.log.debug(f"Ignoring duplicate reaction by {sender.pk} to {item.item_id}")
  1516. return
  1517. intent = sender.intent_for(self)
  1518. if remove:
  1519. await existing.delete()
  1520. await intent.redact(self.mxid, existing.mxid)
  1521. self.log.debug(
  1522. f"Removed {sender.pk}'s reaction to {item.item_id} (redacted {existing.mxid})"
  1523. )
  1524. else:
  1525. timestamp = item.new_reaction.timestamp_ms
  1526. reaction_event_id = await intent.react(
  1527. self.mxid, message.mxid, key=emoji, timestamp=timestamp
  1528. )
  1529. await self._upsert_reaction(
  1530. existing, intent, reaction_event_id, message, sender, emoji, timestamp
  1531. )
  1532. self.log.debug(
  1533. f"Handled {sender.pk}'s reaction to {item.item_id} -> {reaction_event_id}"
  1534. )
  1535. async def _handle_instagram_reactions(
  1536. self, message: DBMessage, reactions: list[Reaction]
  1537. ) -> None:
  1538. old_reactions: dict[int, DBReaction]
  1539. old_reactions = {
  1540. reaction.ig_sender: reaction
  1541. for reaction in await DBReaction.get_all_by_item_id(message.item_id, self.receiver)
  1542. }
  1543. for new_reaction in reactions:
  1544. old_reaction = old_reactions.pop(new_reaction.sender_id, None)
  1545. if old_reaction and old_reaction.reaction == new_reaction.emoji:
  1546. continue
  1547. puppet = await p.Puppet.get_by_pk(new_reaction.sender_id)
  1548. intent = puppet.intent_for(self)
  1549. timestamp = int(time.time() * 1000)
  1550. reaction_event_id = await intent.react(
  1551. self.mxid, message.mxid, new_reaction.emoji, timestamp=timestamp
  1552. )
  1553. await self._upsert_reaction(
  1554. old_reaction,
  1555. intent,
  1556. reaction_event_id,
  1557. message,
  1558. puppet,
  1559. new_reaction.emoji,
  1560. timestamp,
  1561. )
  1562. for old_reaction in old_reactions.values():
  1563. await old_reaction.delete()
  1564. puppet = await p.Puppet.get_by_pk(old_reaction.ig_sender)
  1565. await puppet.intent_for(self).redact(self.mxid, old_reaction.mxid)
  1566. async def handle_instagram_update(self, item: MessageSyncMessage) -> None:
  1567. message = await DBMessage.get_by_item_id(item.item_id, self.receiver)
  1568. if not message:
  1569. return
  1570. if item.has_seen:
  1571. puppet = await p.Puppet.get_by_pk(item.has_seen, create=False)
  1572. if puppet:
  1573. await puppet.intent_for(self).mark_read(self.mxid, message.mxid)
  1574. else:
  1575. async with self._reaction_lock:
  1576. await self._handle_instagram_reactions(
  1577. message, (item.reactions.emojis if item.reactions else [])
  1578. )
  1579. # endregion
  1580. # region Updating portal info
  1581. def _get_thread_name(self, thread: Thread) -> str:
  1582. if self.is_direct:
  1583. if self.other_user_pk == thread.viewer_id and len(thread.users) == 0:
  1584. return "Instagram chat with yourself"
  1585. elif len(thread.users) == 1:
  1586. tpl = self.config["bridge.private_chat_name_template"]
  1587. ui = thread.users[0]
  1588. return tpl.format(
  1589. displayname=ui.full_name or ui.username, id=ui.pk, username=ui.username
  1590. )
  1591. elif thread.thread_title:
  1592. return self.config["bridge.group_chat_name_template"].format(name=thread.thread_title)
  1593. return ""
  1594. async def _get_thread_avatar(self, source: u.User, thread: Thread) -> Optional[ContentURI]:
  1595. if self.is_direct or not thread.thread_image:
  1596. return None
  1597. if self.thread_image_id == thread.thread_image.id:
  1598. return self.avatar_url
  1599. best: Optional[ThreadImageCandidate] = None
  1600. for candidate in thread.thread_image.image_versions2.candidates:
  1601. if best is None or candidate.width > best.width:
  1602. best = candidate
  1603. if not best:
  1604. return None
  1605. data, mimetype = await self._download_instagram_file(source, best.url)
  1606. if not data:
  1607. return None
  1608. mxc = await self.main_intent.upload_media(
  1609. data=data,
  1610. mime_type=mimetype,
  1611. filename=thread.thread_image.id,
  1612. async_upload=self.config["homeserver.async_media"],
  1613. )
  1614. self.thread_image_id = thread.thread_image.id
  1615. return mxc
  1616. async def update_info(self, thread: Thread, source: u.User) -> None:
  1617. changed = await self._update_name(self._get_thread_name(thread))
  1618. if thread_avatar := await self._get_thread_avatar(source, thread):
  1619. changed = await self._update_photo(thread_avatar)
  1620. changed = await self._update_participants(thread.users, source) or changed
  1621. if changed:
  1622. await self.update_bridge_info()
  1623. await self.update()
  1624. # TODO update power levels with thread.admin_user_ids
  1625. async def update_info_from_puppet(self, puppet: p.Puppet | None = None) -> None:
  1626. if not self.is_direct:
  1627. return
  1628. if not puppet:
  1629. puppet = await self.get_dm_puppet()
  1630. await self._update_photo_from_puppet(puppet)
  1631. if self.name and not self.name_set:
  1632. await self._update_name(self.name)
  1633. async def _update_name(self, name: str) -> bool:
  1634. if name and (self.name != name or not self.name_set):
  1635. self.name = name
  1636. if self.mxid:
  1637. try:
  1638. await self.main_intent.set_room_name(self.mxid, name)
  1639. self.name_set = True
  1640. except Exception:
  1641. self.log.exception("Failed to update name")
  1642. self.name_set = False
  1643. return True
  1644. return False
  1645. async def _update_photo_from_puppet(self, puppet: p.Puppet) -> bool:
  1646. if not self.private_chat_portal_meta and not self.encrypted:
  1647. return False
  1648. return await self._update_photo(puppet.photo_mxc)
  1649. async def _update_photo(self, photo_mxc: ContentURI) -> bool:
  1650. if self.avatar_set and self.avatar_url == photo_mxc:
  1651. return False
  1652. self.avatar_url = photo_mxc
  1653. if self.mxid:
  1654. try:
  1655. await self.main_intent.set_room_avatar(self.mxid, photo_mxc)
  1656. self.avatar_set = True
  1657. except Exception:
  1658. self.log.exception("Failed to set room avatar")
  1659. self.avatar_set = False
  1660. return True
  1661. async def _update_participants(self, users: list[ThreadUser], source: u.User) -> bool:
  1662. meta_changed = False
  1663. # Make sure puppets who should be here are here
  1664. for user in users:
  1665. puppet = await p.Puppet.get_by_pk(user.pk)
  1666. await puppet.update_info(user, source)
  1667. if self.mxid:
  1668. await puppet.intent_for(self).ensure_joined(self.mxid)
  1669. if puppet.pk == self.other_user_pk:
  1670. meta_changed = await self._update_photo_from_puppet(puppet)
  1671. if self.mxid:
  1672. # Kick puppets who shouldn't be here
  1673. current_members = {int(user.pk) for user in users}
  1674. for user_id in await self.main_intent.get_room_members(self.mxid):
  1675. pk = p.Puppet.get_id_from_mxid(user_id)
  1676. if pk and pk not in current_members and pk != self.other_user_pk:
  1677. await self.main_intent.kick_user(
  1678. self.mxid,
  1679. p.Puppet.get_mxid_from_id(pk),
  1680. reason="User had left this Instagram DM",
  1681. )
  1682. return meta_changed
  1683. async def _update_read_receipts(self, receipts: dict[int | str, ThreadUserLastSeenAt]) -> None:
  1684. for user_id, receipt in receipts.items():
  1685. message: DBMessage | DBReaction
  1686. message = await DBMessage.get_by_item_id(receipt.item_id, self.receiver)
  1687. if not message:
  1688. reaction: DBReaction
  1689. message, reaction = await asyncio.gather(
  1690. DBMessage.get_closest(self.mxid, int(receipt.timestamp)),
  1691. DBReaction.get_closest(self.mxid, receipt.timestamp_ms),
  1692. )
  1693. if (not message or not message.mxid) and not reaction:
  1694. self.log.debug(
  1695. "Couldn't find message %s to mark as read by %s", receipt, user_id
  1696. )
  1697. continue
  1698. elif not message or (reaction and reaction.mx_timestamp > message.ig_timestamp_ms):
  1699. message = reaction
  1700. puppet = await p.Puppet.get_by_pk(int(user_id), create=False)
  1701. if not puppet:
  1702. continue
  1703. try:
  1704. await puppet.intent_for(self).mark_read(message.mx_room, message.mxid)
  1705. except Exception:
  1706. self.log.warning(
  1707. f"Failed to mark {message.mxid} in {message.mx_room} "
  1708. f"as read by {puppet.intent.mxid}",
  1709. exc_info=True,
  1710. )
  1711. async def get_dm_puppet(self) -> p.Puppet | None:
  1712. if not self.is_direct:
  1713. return None
  1714. return await p.Puppet.get_by_pk(self.other_user_pk)
  1715. # endregion
  1716. # region Backfill
  1717. async def enqueue_immediate_backfill(self, source: u.User, priority: int) -> None:
  1718. assert self.config["bridge.backfill.msc2716"]
  1719. if not await Backfill.get(source.mxid, self.thread_id, self.receiver):
  1720. await Backfill.new(
  1721. source.mxid,
  1722. priority,
  1723. self.thread_id,
  1724. self.receiver,
  1725. self.config["bridge.backfill.incremental.max_pages"],
  1726. self.config["bridge.backfill.incremental.page_delay"],
  1727. self.config["bridge.backfill.incremental.post_batch_delay"],
  1728. self.config["bridge.backfill.incremental.max_total_pages"],
  1729. ).insert()
  1730. async def backfill(self, source: u.User, backfill_request: Backfill) -> None:
  1731. try:
  1732. last_message_ig_timestamp = await self._backfill(source, backfill_request)
  1733. if (
  1734. last_message_ig_timestamp is not None
  1735. and not self.bridge.homeserver_software.is_hungry
  1736. and self.config["bridge.backfill.msc2716"]
  1737. ):
  1738. await self.send_post_backfill_dummy(last_message_ig_timestamp)
  1739. finally:
  1740. # Always sleep after the backfill request is finished processing, even if it errors.
  1741. await asyncio.sleep(backfill_request.post_batch_delay)
  1742. async def _backfill(self, source: u.User, backfill_request: Backfill) -> int | None:
  1743. assert source.client
  1744. self.log.debug("Backfill request: %s", backfill_request)
  1745. num_pages = backfill_request.num_pages
  1746. self.log.debug(
  1747. "Backfilling up to %d pages of history in %s through %s",
  1748. num_pages,
  1749. self.mxid,
  1750. source.mxid,
  1751. )
  1752. try:
  1753. if self.cursor:
  1754. self.log.debug(
  1755. f"There is a cursor for the chat, fetching messages before {self.cursor}"
  1756. )
  1757. resp = await source.client.get_thread(
  1758. self.thread_id, seq_id=source.seq_id, cursor=self.cursor
  1759. )
  1760. else:
  1761. self.log.debug(
  1762. "There is no first message in the chat, starting with the most recent messages"
  1763. )
  1764. resp = await source.client.get_thread(self.thread_id, seq_id=source.seq_id)
  1765. except IGRateLimitError as e:
  1766. backoff = self.config.get("bridge.backfill.backoff.message_history", 300)
  1767. self.log.warning(
  1768. f"Backfilling failed due to rate limit. Waiting for {backoff} seconds before "
  1769. f"resuming. Error: {e}"
  1770. )
  1771. await asyncio.sleep(backoff)
  1772. raise
  1773. async def dedup_messages(messages: list[ThreadItem]) -> list[ThreadItem]:
  1774. deduped = []
  1775. # Sometimes (seems like on Facebook chats) it fetches the first message in the chat over
  1776. # and over again.
  1777. for item in messages:
  1778. # Check in-memory queues for duplicates
  1779. if item.item_id in self._msgid_dedup:
  1780. self.log.debug(
  1781. f"Ignoring message {item.item_id} ({item.client_context}) by {item.user_id}"
  1782. " as it was already handled (message.id in dedup queue)"
  1783. )
  1784. continue
  1785. self._msgid_dedup.appendleft(item.item_id)
  1786. # Check database for duplicates
  1787. if await DBMessage.get_by_item_id(item.item_id, self.receiver) is not None:
  1788. self.log.debug(
  1789. f"Ignoring message {item.item_id} ({item.client_context}) by {item.user_id}"
  1790. " as it was already handled (message.id in database)"
  1791. )
  1792. continue
  1793. deduped.append(item)
  1794. return deduped
  1795. messages = await dedup_messages(resp.thread.items)
  1796. cursor = resp.thread.oldest_cursor
  1797. backfill_more = resp.thread.has_older
  1798. if len(messages) == 0:
  1799. self.log.debug("No messages to backfill.")
  1800. return None
  1801. last_message_timestamp = messages[-1].timestamp_ms
  1802. pages_to_backfill = backfill_request.num_pages
  1803. if backfill_request.max_total_pages > -1:
  1804. pages_to_backfill = min(pages_to_backfill, backfill_request.max_total_pages)
  1805. pages_backfilled = 0
  1806. for i in range(pages_to_backfill):
  1807. base_insertion_event_id = await self.backfill_message_page(
  1808. source, list(reversed(messages))
  1809. )
  1810. self.cursor = cursor
  1811. await self.save()
  1812. pages_backfilled += 1
  1813. if base_insertion_event_id:
  1814. self.historical_base_insertion_event_id = base_insertion_event_id
  1815. await self.save()
  1816. if backfill_more and i < pages_to_backfill - 1:
  1817. # Sleep before fetching another page of messages.
  1818. await asyncio.sleep(backfill_request.page_delay)
  1819. # Fetch more messages
  1820. try:
  1821. resp = await source.client.get_thread(
  1822. self.thread_id, seq_id=source.seq_id, cursor=self.cursor
  1823. )
  1824. messages = await dedup_messages(resp.thread.items)
  1825. cursor = resp.thread.oldest_cursor
  1826. backfill_more &= resp.thread.has_older
  1827. except IGRateLimitError as e:
  1828. backoff = self.config.get("bridge.backfill.backoff.message_history", 300)
  1829. self.log.warning(
  1830. f"Backfilling failed due to rate limit. Waiting for {backoff} seconds "
  1831. "before resuming."
  1832. )
  1833. await asyncio.sleep(backoff)
  1834. # If we hit the rate limit, then we will want to give up for now, but enqueue
  1835. # additional backfill to do later.
  1836. break
  1837. if backfill_request.max_total_pages == -1:
  1838. new_max_total_pages = -1
  1839. else:
  1840. new_max_total_pages = backfill_request.max_total_pages - pages_backfilled
  1841. if new_max_total_pages <= 0:
  1842. backfill_more = False
  1843. if backfill_more:
  1844. self.log.debug("Enqueueing more backfill")
  1845. await Backfill.new(
  1846. source.mxid,
  1847. # Always enqueue subsequent backfills at the lowest priority
  1848. 2,
  1849. self.thread_id,
  1850. self.receiver,
  1851. backfill_request.num_pages,
  1852. backfill_request.page_delay,
  1853. backfill_request.post_batch_delay,
  1854. new_max_total_pages,
  1855. ).insert()
  1856. else:
  1857. self.log.debug("No more messages to backfill")
  1858. await self._update_read_receipts(resp.thread.last_seen_at)
  1859. return last_message_timestamp
  1860. async def backfill_message_page(
  1861. self,
  1862. source: u.User,
  1863. message_page: list[ThreadItem],
  1864. forward: bool = False,
  1865. last_message: DBMessage | None = None,
  1866. mark_read: bool = False,
  1867. ) -> EventID | None:
  1868. """
  1869. Backfills a page of messages to Matrix. The messages should be in order from oldest to
  1870. newest.
  1871. Returns: a tuple containing the number of messages that were actually bridged, the
  1872. timestamp of the oldest bridged message and the base insertion event ID if it exists.
  1873. """
  1874. assert source.client
  1875. if len(message_page) == 0:
  1876. return None
  1877. if forward:
  1878. assert (last_message and last_message.mxid) or self.first_event_id
  1879. prev_event_id = last_message.mxid if last_message else self.first_event_id
  1880. else:
  1881. assert self.config["bridge.backfill.msc2716"]
  1882. assert self.first_event_id
  1883. prev_event_id = self.first_event_id
  1884. assert self.mxid
  1885. oldest_message_in_page = message_page[0]
  1886. oldest_msg_timestamp = oldest_message_in_page.timestamp_ms
  1887. batch_messages: list[BatchSendEvent] = []
  1888. state_events_at_start: list[BatchSendStateEvent] = []
  1889. added_members = set()
  1890. current_members = await self.main_intent.state_store.get_members(
  1891. self.mxid, memberships=(Membership.JOIN,)
  1892. )
  1893. def add_member(puppet: p.Puppet, mxid: UserID):
  1894. assert self.mxid
  1895. if mxid in added_members:
  1896. return
  1897. if (
  1898. self.bridge.homeserver_software.is_hungry
  1899. or not self.config["bridge.backfill.msc2716"]
  1900. ):
  1901. # Hungryserv doesn't expect or check state events at start.
  1902. added_members.add(mxid)
  1903. return
  1904. content_args = {"avatar_url": puppet.photo_mxc, "displayname": puppet.name}
  1905. state_events_at_start.extend(
  1906. [
  1907. BatchSendStateEvent(
  1908. content=MemberStateEventContent(Membership.INVITE, **content_args),
  1909. type=EventType.ROOM_MEMBER,
  1910. sender=self.main_intent.mxid,
  1911. state_key=mxid,
  1912. timestamp=oldest_msg_timestamp,
  1913. ),
  1914. BatchSendStateEvent(
  1915. content=MemberStateEventContent(Membership.JOIN, **content_args),
  1916. type=EventType.ROOM_MEMBER,
  1917. sender=mxid,
  1918. state_key=mxid,
  1919. timestamp=oldest_msg_timestamp,
  1920. ),
  1921. ]
  1922. )
  1923. added_members.add(mxid)
  1924. async def intent_for(user_id: int) -> tuple[p.Puppet, IntentAPI]:
  1925. puppet: p.Puppet = await p.Puppet.get_by_pk(user_id)
  1926. if puppet:
  1927. intent = puppet.intent_for(self)
  1928. else:
  1929. intent = self.main_intent
  1930. if puppet.is_real_user and not self._can_double_puppet_backfill(intent.mxid):
  1931. intent = puppet.default_mxid_intent
  1932. return puppet, intent
  1933. message_infos: list[tuple[ThreadItem | Reaction, int]] = []
  1934. intents: list[IntentAPI] = []
  1935. for message in message_page:
  1936. puppet, intent = await intent_for(message.user_id)
  1937. # Convert the message
  1938. converted = await self.convert_instagram_item(source, puppet, message)
  1939. if not converted:
  1940. self.log.debug(f"Skipping unsupported message in backfill {message.item_id}")
  1941. continue
  1942. if intent.mxid not in current_members:
  1943. add_member(puppet, intent.mxid)
  1944. d_event_id = None
  1945. for index, (event_type, content) in enumerate(converted):
  1946. if self.encrypted and self.matrix.e2ee:
  1947. event_type, content = await self.matrix.e2ee.encrypt(
  1948. self.mxid, event_type, content
  1949. )
  1950. if intent.api.is_real_user and intent.api.bridge_name is not None:
  1951. content[DOUBLE_PUPPET_SOURCE_KEY] = intent.api.bridge_name
  1952. if self.bridge.homeserver_software.is_hungry:
  1953. d_event_id = self._deterministic_event_id(puppet, message.item_id, index)
  1954. message_infos.append((message, index))
  1955. batch_messages.append(
  1956. BatchSendEvent(
  1957. content=content,
  1958. type=event_type,
  1959. sender=intent.mxid,
  1960. timestamp=message.timestamp_ms,
  1961. event_id=d_event_id,
  1962. )
  1963. )
  1964. intents.append(intent)
  1965. if self.bridge.homeserver_software.is_hungry and message.reactions:
  1966. for reaction in message.reactions.emojis:
  1967. puppet, intent = await intent_for(reaction.sender_id)
  1968. reaction_event = ReactionEventContent()
  1969. reaction_event.relates_to = RelatesTo(
  1970. rel_type=RelationType.ANNOTATION, event_id=d_event_id, key=reaction.emoji
  1971. )
  1972. if intent.api.is_real_user and intent.api.bridge_name is not None:
  1973. reaction_event[DOUBLE_PUPPET_SOURCE_KEY] = intent.api.bridge_name
  1974. message_infos.append((reaction, 0))
  1975. batch_messages.append(
  1976. BatchSendEvent(
  1977. content=reaction_event,
  1978. type=EventType.REACTION,
  1979. sender=intent.mxid,
  1980. timestamp=reaction.timestamp_ms or message.timestamp_ms,
  1981. )
  1982. )
  1983. if not batch_messages:
  1984. return None
  1985. if (
  1986. not self.bridge.homeserver_software.is_hungry
  1987. and self.config["bridge.backfill.msc2716"]
  1988. and (forward or self.next_batch_id is None)
  1989. ):
  1990. self.log.debug("Sending dummy event to avoid forward extremity errors")
  1991. await self.az.intent.send_message_event(
  1992. self.mxid, EventType("fi.mau.dummy.pre_backfill", EventType.Class.MESSAGE), {}
  1993. )
  1994. self.log.info(
  1995. "Sending %d %s messages to %s with batch ID %s and previous event ID %s",
  1996. len(batch_messages),
  1997. "new" if forward else "historical",
  1998. self.mxid,
  1999. self.next_batch_id,
  2000. prev_event_id,
  2001. )
  2002. if self.bridge.homeserver_software.is_hungry:
  2003. self.log.debug("Batch message event IDs %s", [m.event_id for m in batch_messages])
  2004. base_insertion_event_id = None
  2005. if self.config["bridge.backfill.msc2716"]:
  2006. batch_send_resp = await self.main_intent.batch_send(
  2007. self.mxid,
  2008. prev_event_id,
  2009. batch_id=self.next_batch_id,
  2010. events=batch_messages,
  2011. state_events_at_start=state_events_at_start,
  2012. beeper_new_messages=forward,
  2013. beeper_mark_read_by=source.mxid if mark_read else None,
  2014. )
  2015. base_insertion_event_id = batch_send_resp.base_insertion_event_id
  2016. event_ids = batch_send_resp.event_ids
  2017. else:
  2018. batch_send_resp = None
  2019. event_ids = [
  2020. await intent.send_message_event(
  2021. self.mxid, evt.type, evt.content, timestamp=evt.timestamp
  2022. )
  2023. for evt, intent in zip(reversed(batch_messages), reversed(intents))
  2024. ]
  2025. await self._finish_batch(event_ids, message_infos)
  2026. if not forward:
  2027. assert batch_send_resp
  2028. self.log.debug("Got next batch ID %s for %s", batch_send_resp.next_batch_id, self.mxid)
  2029. self.next_batch_id = batch_send_resp.next_batch_id
  2030. await self.save()
  2031. return base_insertion_event_id
  2032. def _can_double_puppet_backfill(self, custom_mxid: UserID) -> bool:
  2033. return self.config["bridge.backfill.double_puppet_backfill"] and (
  2034. # Hungryserv can batch send any users
  2035. self.bridge.homeserver_software.is_hungry
  2036. # Non-MSC2716 backfill can use any double puppet
  2037. or not self.config["bridge.backfill.msc2716"]
  2038. # Local users can be double puppeted even with MSC2716
  2039. or (custom_mxid[custom_mxid.index(":") + 1 :] == self.config["homeserver.domain"])
  2040. )
  2041. async def _finish_batch(
  2042. self, event_ids: list[EventID], message_infos: list[tuple[ThreadItem | Reaction, int]]
  2043. ):
  2044. # We have to do this slightly annoying processing of the event IDs and message infos so
  2045. # that we only map the last event ID to the message.
  2046. # When inline captions are enabled, this will have no effect since index will always be 0
  2047. # since there's only ever one event per message.
  2048. current_message = None
  2049. messages = []
  2050. reactions = []
  2051. message_id = None
  2052. for event_id, (message_or_reaction, index) in zip(event_ids, message_infos):
  2053. if isinstance(message_or_reaction, ThreadItem):
  2054. message = message_or_reaction
  2055. if index == 0 and current_message:
  2056. # This means that all of the events for the previous message have been processed,
  2057. # and the current_message is the most recent event for that message.
  2058. messages.append(current_message)
  2059. current_message = DBMessage(
  2060. mxid=event_id,
  2061. mx_room=self.mxid,
  2062. item_id=message.item_id,
  2063. client_context=message.client_context,
  2064. receiver=self.receiver,
  2065. sender=message.user_id,
  2066. ig_timestamp=message.timestamp,
  2067. )
  2068. message_id = message.item_id
  2069. else:
  2070. assert message_id
  2071. reaction = message_or_reaction
  2072. reactions.append(
  2073. DBReaction(
  2074. mxid=event_id,
  2075. mx_room=self.mxid,
  2076. ig_item_id=message_id,
  2077. ig_receiver=self.receiver,
  2078. ig_sender=reaction.sender_id,
  2079. reaction=reaction.emoji,
  2080. mx_timestamp=reaction.timestamp_ms,
  2081. )
  2082. )
  2083. if current_message:
  2084. messages.append(current_message)
  2085. try:
  2086. await DBMessage.bulk_insert(messages)
  2087. except Exception:
  2088. self.log.exception("Failed to store batch message IDs")
  2089. try:
  2090. for reaction in reactions:
  2091. await reaction.insert()
  2092. except Exception:
  2093. self.log.exception("Failed to store backfilled reactions")
  2094. async def send_post_backfill_dummy(
  2095. self,
  2096. last_message_ig_timestamp: int,
  2097. base_insertion_event_id: EventID | None = None,
  2098. ):
  2099. if not self.config["bridge.backfill.msc2716"]:
  2100. return
  2101. assert self.mxid
  2102. if not base_insertion_event_id:
  2103. base_insertion_event_id = self.historical_base_insertion_event_id
  2104. if not base_insertion_event_id:
  2105. self.log.debug(
  2106. "No base insertion event ID in database or from batch send response. Not sending"
  2107. " dummy event."
  2108. )
  2109. return
  2110. event_id = await self.main_intent.send_message_event(
  2111. self.mxid,
  2112. event_type=HistorySyncMarkerMessage,
  2113. content={
  2114. "org.matrix.msc2716.marker.insertion": base_insertion_event_id,
  2115. "m.marker.insertion": base_insertion_event_id,
  2116. },
  2117. )
  2118. await DBMessage(
  2119. mxid=event_id,
  2120. mx_room=self.mxid,
  2121. item_id=f"fi.mau.instagram.post_backfill_dummy.{last_message_ig_timestamp}",
  2122. client_context=None,
  2123. receiver=self.receiver,
  2124. sender=0,
  2125. ig_timestamp=last_message_ig_timestamp,
  2126. ).insert()
  2127. # endregion
  2128. # region Bridge info state event
  2129. @property
  2130. def bridge_info_state_key(self) -> str:
  2131. return f"net.maunium.instagram://instagram/{self.thread_id}"
  2132. @property
  2133. def bridge_info(self) -> dict[str, Any]:
  2134. return {
  2135. "bridgebot": self.az.bot_mxid,
  2136. "creator": self.main_intent.mxid,
  2137. "protocol": {
  2138. "id": "instagram",
  2139. "displayname": "Instagram DM",
  2140. "avatar_url": self.config["appservice.bot_avatar"],
  2141. },
  2142. "channel": {
  2143. "id": self.thread_id,
  2144. "displayname": self.name,
  2145. "avatar_url": self.avatar_url,
  2146. },
  2147. }
  2148. async def update_bridge_info(self) -> None:
  2149. if not self.mxid:
  2150. self.log.debug("Not updating bridge info: no Matrix room created")
  2151. return
  2152. try:
  2153. self.log.debug("Updating bridge info...")
  2154. await self.main_intent.send_state_event(
  2155. self.mxid, StateBridge, self.bridge_info, self.bridge_info_state_key
  2156. )
  2157. # TODO remove this once https://github.com/matrix-org/matrix-doc/pull/2346 is in spec
  2158. await self.main_intent.send_state_event(
  2159. self.mxid, StateHalfShotBridge, self.bridge_info, self.bridge_info_state_key
  2160. )
  2161. except Exception:
  2162. self.log.warning("Failed to update bridge info", exc_info=True)
  2163. # endregion
  2164. # region Creating Matrix rooms
  2165. async def create_matrix_room(self, source: u.User, info: Thread) -> RoomID | None:
  2166. if self.mxid:
  2167. try:
  2168. await self.update_matrix_room(source, info)
  2169. except Exception:
  2170. self.log.exception("Failed to update portal")
  2171. return self.mxid
  2172. async with self._create_room_lock:
  2173. try:
  2174. return await self._create_matrix_room(source, info)
  2175. except Exception:
  2176. self.log.exception("Failed to create portal")
  2177. return None
  2178. def _get_invite_content(self, double_puppet: p.Puppet | None) -> dict[str, bool]:
  2179. invite_content = {}
  2180. if double_puppet:
  2181. invite_content["fi.mau.will_auto_accept"] = True
  2182. if self.is_direct:
  2183. invite_content["is_direct"] = True
  2184. return invite_content
  2185. async def update_matrix_room(self, source: u.User, info: Thread) -> None:
  2186. puppet = await p.Puppet.get_by_custom_mxid(source.mxid)
  2187. await self.main_intent.invite_user(
  2188. self.mxid,
  2189. source.mxid,
  2190. check_cache=True,
  2191. extra_content=self._get_invite_content(puppet),
  2192. )
  2193. if puppet:
  2194. did_join = await puppet.intent.ensure_joined(self.mxid)
  2195. if did_join and self.is_direct:
  2196. await source.update_direct_chats({self.main_intent.mxid: [self.mxid]})
  2197. await self.update_info(info, source)
  2198. await self._update_read_receipts(info.last_seen_at)
  2199. async def _create_matrix_room(self, source: u.User, info: Thread) -> RoomID | None:
  2200. if self.mxid:
  2201. await self.update_matrix_room(source, info)
  2202. return self.mxid
  2203. await self.update_info(info, source)
  2204. self.log.debug("Creating Matrix room")
  2205. name: str | None = None
  2206. initial_state = [
  2207. {
  2208. "type": str(StateBridge),
  2209. "state_key": self.bridge_info_state_key,
  2210. "content": self.bridge_info,
  2211. },
  2212. # TODO remove this once https://github.com/matrix-org/matrix-doc/pull/2346 is in spec
  2213. {
  2214. "type": str(StateHalfShotBridge),
  2215. "state_key": self.bridge_info_state_key,
  2216. "content": self.bridge_info,
  2217. },
  2218. ]
  2219. invites = []
  2220. if self.config["bridge.encryption.default"] and self.matrix.e2ee:
  2221. self.encrypted = True
  2222. initial_state.append(
  2223. {
  2224. "type": "m.room.encryption",
  2225. "content": self.get_encryption_state_event_json(),
  2226. }
  2227. )
  2228. if self.is_direct:
  2229. invites.append(self.az.bot_mxid)
  2230. if self.encrypted or self.private_chat_portal_meta or not self.is_direct:
  2231. name = self.name
  2232. creation_content = {}
  2233. if not self.config["bridge.federate_rooms"]:
  2234. creation_content["m.federate"] = False
  2235. self.mxid = await self.main_intent.create_room(
  2236. name=name,
  2237. is_direct=self.is_direct,
  2238. initial_state=initial_state,
  2239. invitees=invites,
  2240. creation_content=creation_content,
  2241. )
  2242. if not self.mxid:
  2243. raise Exception("Failed to create room: no mxid returned")
  2244. if self.encrypted and self.matrix.e2ee and self.is_direct:
  2245. try:
  2246. await self.az.intent.ensure_joined(self.mxid)
  2247. except Exception:
  2248. self.log.warning(f"Failed to add bridge bot to new private chat {self.mxid}")
  2249. await self.update()
  2250. self.log.debug(f"Matrix room created: {self.mxid}")
  2251. self.by_mxid[self.mxid] = self
  2252. puppet = await p.Puppet.get_by_custom_mxid(source.mxid)
  2253. await self.main_intent.invite_user(
  2254. self.mxid, source.mxid, extra_content=self._get_invite_content(puppet)
  2255. )
  2256. if puppet:
  2257. try:
  2258. if self.is_direct:
  2259. await source.update_direct_chats({self.main_intent.mxid: [self.mxid]})
  2260. await puppet.intent.join_room_by_id(self.mxid)
  2261. except MatrixError:
  2262. self.log.debug(
  2263. "Failed to join custom puppet into newly created portal", exc_info=True
  2264. )
  2265. await self._update_participants(info.users, source)
  2266. self.log.trace("Sending portal post-create dummy event")
  2267. self.first_event_id = await self.az.intent.send_message_event(
  2268. self.mxid, PortalCreateDummy, {}
  2269. )
  2270. await self.update()
  2271. return self.mxid
  2272. # endregion
  2273. # region Database getters
  2274. async def postinit(self) -> None:
  2275. self.by_thread_id[(self.thread_id, self.receiver)] = self
  2276. if self.mxid:
  2277. self.by_mxid[self.mxid] = self
  2278. self._main_intent = (
  2279. (await p.Puppet.get_by_pk(self.other_user_pk)).default_mxid_intent
  2280. if self.other_user_pk
  2281. else self.az.intent
  2282. )
  2283. async def delete(self) -> None:
  2284. await DBMessage.delete_all(self.mxid)
  2285. self.by_mxid.pop(self.mxid, None)
  2286. self.mxid = None
  2287. self.encrypted = False
  2288. await self.update()
  2289. async def save(self) -> None:
  2290. await self.update()
  2291. @classmethod
  2292. def all_with_room(cls) -> AsyncGenerator[Portal, None]:
  2293. return cls._db_to_portals(super().all_with_room())
  2294. @classmethod
  2295. def find_private_chats_with(cls, other_user: int) -> AsyncGenerator[Portal, None]:
  2296. return cls._db_to_portals(super().find_private_chats_with(other_user))
  2297. @classmethod
  2298. async def find_private_chat(cls, receiver: int, other_user: int) -> Portal | None:
  2299. thread_id = await super().find_private_chat_id(receiver, other_user)
  2300. if not thread_id:
  2301. return None
  2302. return await cls.get_by_thread_id(thread_id, receiver=receiver, is_group=False)
  2303. @classmethod
  2304. async def _db_to_portals(cls, query: Awaitable[list[Portal]]) -> AsyncGenerator[Portal, None]:
  2305. portals = await query
  2306. for index, portal in enumerate(portals):
  2307. try:
  2308. yield cls.by_thread_id[(portal.thread_id, portal.receiver)]
  2309. except KeyError:
  2310. await portal.postinit()
  2311. yield portal
  2312. @classmethod
  2313. @async_getter_lock
  2314. async def get_by_mxid(cls, mxid: RoomID) -> Portal | None:
  2315. try:
  2316. return cls.by_mxid[mxid]
  2317. except KeyError:
  2318. pass
  2319. portal = cast(cls, await super().get_by_mxid(mxid))
  2320. if portal is not None:
  2321. await portal.postinit()
  2322. return portal
  2323. return None
  2324. @classmethod
  2325. @async_getter_lock
  2326. async def get_by_thread_id(
  2327. cls,
  2328. thread_id: str,
  2329. *,
  2330. receiver: int,
  2331. is_group: bool | None = None,
  2332. other_user_pk: int | None = None,
  2333. ) -> Portal | None:
  2334. if is_group and receiver != 0:
  2335. receiver = 0
  2336. try:
  2337. return cls.by_thread_id[(thread_id, receiver)]
  2338. except KeyError:
  2339. pass
  2340. if is_group is None and receiver != 0:
  2341. try:
  2342. return cls.by_thread_id[(thread_id, 0)]
  2343. except KeyError:
  2344. pass
  2345. portal = cast(
  2346. cls,
  2347. await super().get_by_thread_id(
  2348. thread_id, receiver=receiver, rec_must_match=is_group is not None
  2349. ),
  2350. )
  2351. if portal is not None:
  2352. await portal.postinit()
  2353. return portal
  2354. if is_group is not None:
  2355. portal = cls(thread_id, receiver, other_user_pk=other_user_pk)
  2356. await portal.insert()
  2357. await portal.postinit()
  2358. return portal
  2359. return None
  2360. @classmethod
  2361. async def get_by_thread(cls, thread: Thread, receiver: int) -> Portal | None:
  2362. if thread.is_group:
  2363. receiver = 0
  2364. other_user_pk = None
  2365. else:
  2366. if len(thread.users) == 0:
  2367. other_user_pk = receiver
  2368. else:
  2369. other_user_pk = thread.users[0].pk
  2370. return await cls.get_by_thread_id(
  2371. thread.thread_id,
  2372. receiver=receiver,
  2373. is_group=thread.is_group,
  2374. other_user_pk=other_user_pk,
  2375. )
  2376. # endregion