user.py 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739
  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, AsyncGenerator, AsyncIterable, Awaitable, cast
  18. import asyncio
  19. import logging
  20. import time
  21. from mauigpapi import AndroidAPI, AndroidMQTT, AndroidState
  22. from mauigpapi.errors import (
  23. IGChallengeError,
  24. IGCheckpointError,
  25. IGConsentRequiredError,
  26. IGNotLoggedInError,
  27. IGRateLimitError,
  28. IGUserIDNotFoundError,
  29. IrisSubscribeError,
  30. MQTTNotConnected,
  31. MQTTNotLoggedIn,
  32. )
  33. from mauigpapi.mqtt import (
  34. Connect,
  35. Disconnect,
  36. GraphQLSubscription,
  37. NewSequenceID,
  38. SkywalkerSubscription,
  39. )
  40. from mauigpapi.types import (
  41. ActivityIndicatorData,
  42. CurrentUser,
  43. MessageSyncEvent,
  44. Operation,
  45. RealtimeDirectEvent,
  46. Thread,
  47. ThreadSyncEvent,
  48. TypingStatus,
  49. )
  50. from mautrix.appservice import AppService
  51. from mautrix.bridge import BaseUser, async_getter_lock
  52. from mautrix.types import EventID, MessageType, RoomID, TextMessageEventContent, UserID
  53. from mautrix.util.bridge_state import BridgeState, BridgeStateEvent
  54. from mautrix.util.logging import TraceLogger
  55. from mautrix.util.opt_prometheus import Gauge, Summary, async_time
  56. from . import portal as po, puppet as pu
  57. from .config import Config
  58. from .db import Portal as DBPortal, User as DBUser
  59. if TYPE_CHECKING:
  60. from .__main__ import InstagramBridge
  61. METRIC_MESSAGE = Summary("bridge_on_message", "calls to handle_message")
  62. METRIC_THREAD_SYNC = Summary("bridge_on_thread_sync", "calls to handle_thread_sync")
  63. METRIC_RTD = Summary("bridge_on_rtd", "calls to handle_rtd")
  64. METRIC_LOGGED_IN = Gauge("bridge_logged_in", "Users logged into the bridge")
  65. METRIC_CONNECTED = Gauge("bridge_connected", "Bridged users connected to Instagram")
  66. BridgeState.human_readable_errors.update(
  67. {
  68. "ig-connection-error": "Instagram disconnected unexpectedly",
  69. "ig-auth-error": "Authentication error from Instagram: {message}",
  70. "ig-checkpoint": "Instagram checkpoint error. Please check the Instagram website.",
  71. "ig-consent-required": "Instagram requires a consent update. Please check the Instagram website.",
  72. "ig-checkpoint-locked": "Instagram checkpoint error. Please check the Instagram website.",
  73. "ig-rate-limit": "Got Instagram ratelimit error, waiting a few minutes before retrying...",
  74. "ig-disconnected": None,
  75. "ig-no-mqtt": "You're not connected to Instagram",
  76. "logged-out": "You're not logged into Instagram",
  77. }
  78. )
  79. class User(DBUser, BaseUser):
  80. ig_base_log: TraceLogger = logging.getLogger("mau.instagram")
  81. _activity_indicator_ids: dict[str, int] = {}
  82. by_mxid: dict[UserID, User] = {}
  83. by_igpk: dict[int, User] = {}
  84. config: Config
  85. az: AppService
  86. loop: asyncio.AbstractEventLoop
  87. client: AndroidAPI | None
  88. mqtt: AndroidMQTT | None
  89. _listen_task: asyncio.Task | None = None
  90. _seq_id_save_task: asyncio.Task | None
  91. permission_level: str
  92. username: str | None
  93. _notice_room_lock: asyncio.Lock
  94. _notice_send_lock: asyncio.Lock
  95. _is_logged_in: bool
  96. _is_connected: bool
  97. shutdown: bool
  98. remote_typing_status: TypingStatus | None
  99. def __init__(
  100. self,
  101. mxid: UserID,
  102. igpk: int | None = None,
  103. state: AndroidState | None = None,
  104. notice_room: RoomID | None = None,
  105. seq_id: int | None = None,
  106. snapshot_at_ms: int | None = None,
  107. ) -> None:
  108. super().__init__(
  109. mxid=mxid,
  110. igpk=igpk,
  111. state=state,
  112. notice_room=notice_room,
  113. seq_id=seq_id,
  114. snapshot_at_ms=snapshot_at_ms,
  115. )
  116. BaseUser.__init__(self)
  117. self._notice_room_lock = asyncio.Lock()
  118. self._notice_send_lock = asyncio.Lock()
  119. perms = self.config.get_permissions(mxid)
  120. self.relay_whitelisted, self.is_whitelisted, self.is_admin, self.permission_level = perms
  121. self.client = None
  122. self.mqtt = None
  123. self.username = None
  124. self._is_logged_in = False
  125. self._is_connected = False
  126. self._is_refreshing = False
  127. self.shutdown = False
  128. self._listen_task = None
  129. self.remote_typing_status = None
  130. self._seq_id_save_task = None
  131. @classmethod
  132. def init_cls(cls, bridge: "InstagramBridge") -> AsyncIterable[Awaitable[None]]:
  133. cls.bridge = bridge
  134. cls.config = bridge.config
  135. cls.az = bridge.az
  136. cls.loop = bridge.loop
  137. return (user.try_connect() async for user in cls.all_logged_in())
  138. # region Connection management
  139. async def is_logged_in(self) -> bool:
  140. return bool(self.client) and self._is_logged_in
  141. async def get_puppet(self) -> pu.Puppet | None:
  142. if not self.igpk:
  143. return None
  144. return await pu.Puppet.get_by_pk(self.igpk)
  145. async def get_portal_with(self, puppet: pu.Puppet, create: bool = True) -> po.Portal | None:
  146. if not self.igpk:
  147. return None
  148. portal = await po.Portal.find_private_chat(self.igpk, puppet.pk)
  149. if portal:
  150. return portal
  151. if create:
  152. # TODO add error handling somewhere
  153. thread = await self.client.create_group_thread([puppet.pk])
  154. portal = await po.Portal.get_by_thread(thread, self.igpk)
  155. await portal.update_info(thread, self)
  156. return portal
  157. return None
  158. async def try_connect(self) -> None:
  159. try:
  160. await self.connect()
  161. except Exception as e:
  162. self.log.exception("Error while connecting to Instagram")
  163. await self.push_bridge_state(
  164. BridgeStateEvent.UNKNOWN_ERROR, info={"python_error": str(e)}
  165. )
  166. @property
  167. def api_log(self) -> TraceLogger:
  168. return self.ig_base_log.getChild("http").getChild(self.mxid)
  169. @property
  170. def is_connected(self) -> bool:
  171. return bool(self.client) and bool(self.mqtt) and self._is_connected
  172. async def connect(self, user: CurrentUser | None = None) -> None:
  173. if not self.state:
  174. await self.push_bridge_state(BridgeStateEvent.BAD_CREDENTIALS, error="logged-out")
  175. return
  176. client = AndroidAPI(self.state, log=self.api_log)
  177. if not user:
  178. try:
  179. resp = await client.current_user()
  180. user = resp.user
  181. except IGNotLoggedInError as e:
  182. self.log.warning(f"Failed to connect to Instagram: {e}, logging out")
  183. await self.logout(error=e)
  184. return
  185. except (IGChallengeError, IGConsentRequiredError) as e:
  186. await self._handle_checkpoint(e, on="connect", client=client)
  187. return
  188. self.client = client
  189. self._is_logged_in = True
  190. self.igpk = user.pk
  191. self.username = user.username
  192. await self.push_bridge_state(BridgeStateEvent.CONNECTING)
  193. self._track_metric(METRIC_LOGGED_IN, True)
  194. self.by_igpk[self.igpk] = self
  195. self.mqtt = AndroidMQTT(
  196. self.state, loop=self.loop, log=self.ig_base_log.getChild("mqtt").getChild(self.mxid)
  197. )
  198. self.mqtt.add_event_handler(Connect, self.on_connect)
  199. self.mqtt.add_event_handler(Disconnect, self.on_disconnect)
  200. self.mqtt.add_event_handler(NewSequenceID, self.update_seq_id)
  201. self.mqtt.add_event_handler(MessageSyncEvent, self.handle_message)
  202. self.mqtt.add_event_handler(ThreadSyncEvent, self.handle_thread_sync)
  203. self.mqtt.add_event_handler(RealtimeDirectEvent, self.handle_rtd)
  204. await self.update()
  205. self.loop.create_task(self._try_sync_puppet(user))
  206. if not self.seq_id or self.config["bridge.resync_on_startup"]:
  207. self.loop.create_task(self._try_sync())
  208. else:
  209. self.log.debug("Connecting to MQTT directly as resync_on_startup is false")
  210. self.start_listen()
  211. async def on_connect(self, evt: Connect) -> None:
  212. self.log.debug("Connected to Instagram")
  213. self._track_metric(METRIC_CONNECTED, True)
  214. self._is_connected = True
  215. await self.send_bridge_notice("Connected to Instagram")
  216. await self.push_bridge_state(BridgeStateEvent.CONNECTED)
  217. async def on_disconnect(self, evt: Disconnect) -> None:
  218. self.log.debug("Disconnected from Instagram")
  219. self._track_metric(METRIC_CONNECTED, False)
  220. self._is_connected = False
  221. # TODO this stuff could probably be moved to mautrix-python
  222. async def get_notice_room(self) -> RoomID:
  223. if not self.notice_room:
  224. async with self._notice_room_lock:
  225. # If someone already created the room while this call was waiting,
  226. # don't make a new room
  227. if self.notice_room:
  228. return self.notice_room
  229. creation_content = {}
  230. if not self.config["bridge.federate_rooms"]:
  231. creation_content["m.federate"] = False
  232. self.notice_room = await self.az.intent.create_room(
  233. is_direct=True,
  234. invitees=[self.mxid],
  235. topic="Instagram bridge notices",
  236. creation_content=creation_content,
  237. )
  238. await self.update()
  239. return self.notice_room
  240. async def fill_bridge_state(self, state: BridgeState) -> None:
  241. await super().fill_bridge_state(state)
  242. if not state.remote_id:
  243. if self.igpk:
  244. state.remote_id = str(self.igpk)
  245. else:
  246. try:
  247. state.remote_id = self.state.user_id
  248. except IGUserIDNotFoundError:
  249. state.remote_id = None
  250. if self.username:
  251. state.remote_name = f"@{self.username}"
  252. async def get_bridge_states(self) -> list[BridgeState]:
  253. if not self.state:
  254. return []
  255. state = BridgeState(state_event=BridgeStateEvent.UNKNOWN_ERROR)
  256. if self.is_connected:
  257. state.state_event = BridgeStateEvent.CONNECTED
  258. elif self._is_refreshing or self.mqtt:
  259. state.state_event = BridgeStateEvent.TRANSIENT_DISCONNECT
  260. return [state]
  261. async def send_bridge_notice(
  262. self,
  263. text: str,
  264. edit: EventID | None = None,
  265. state_event: BridgeStateEvent | None = None,
  266. important: bool = False,
  267. error_code: str | None = None,
  268. error_message: str | None = None,
  269. info: dict | None = None,
  270. ) -> EventID | None:
  271. if state_event:
  272. await self.push_bridge_state(
  273. state_event,
  274. error=error_code,
  275. message=error_message if error_code else text,
  276. info=info,
  277. )
  278. if self.config["bridge.disable_bridge_notices"]:
  279. return None
  280. if not important and not self.config["bridge.unimportant_bridge_notices"]:
  281. self.log.debug("Not sending unimportant bridge notice: %s", text)
  282. return None
  283. event_id = None
  284. try:
  285. self.log.debug("Sending bridge notice: %s", text)
  286. content = TextMessageEventContent(
  287. body=text, msgtype=(MessageType.TEXT if important else MessageType.NOTICE)
  288. )
  289. if edit:
  290. content.set_edit(edit)
  291. # This is locked to prevent notices going out in the wrong order
  292. async with self._notice_send_lock:
  293. event_id = await self.az.intent.send_message(await self.get_notice_room(), content)
  294. except Exception:
  295. self.log.warning("Failed to send bridge notice", exc_info=True)
  296. return edit or event_id
  297. async def _try_sync_puppet(self, user_info: CurrentUser) -> None:
  298. puppet = await pu.Puppet.get_by_pk(self.igpk)
  299. try:
  300. await puppet.update_info(user_info, self)
  301. except Exception:
  302. self.log.exception("Failed to update own puppet info")
  303. try:
  304. if puppet.custom_mxid != self.mxid and puppet.can_auto_login(self.mxid):
  305. self.log.info(f"Automatically enabling custom puppet")
  306. await puppet.switch_mxid(access_token="auto", mxid=self.mxid)
  307. except Exception:
  308. self.log.exception("Failed to automatically enable custom puppet")
  309. async def _try_sync(self) -> None:
  310. try:
  311. await self.sync()
  312. except Exception as e:
  313. self.log.exception("Exception while syncing")
  314. if isinstance(e, IGCheckpointError):
  315. self.log.debug("Checkpoint error content: %s", e.body)
  316. await self.push_bridge_state(
  317. BridgeStateEvent.UNKNOWN_ERROR, info={"python_error": str(e)}
  318. )
  319. async def get_direct_chats(self) -> dict[UserID, list[RoomID]]:
  320. return {
  321. pu.Puppet.get_mxid_from_id(portal.other_user_pk): [portal.mxid]
  322. for portal in await DBPortal.find_private_chats_of(self.igpk)
  323. if portal.mxid
  324. }
  325. async def refresh(self, resync: bool = True) -> None:
  326. self._is_refreshing = True
  327. try:
  328. await self.stop_listen()
  329. if resync:
  330. retry_count = 0
  331. minutes = 1
  332. while True:
  333. try:
  334. await self.sync()
  335. return
  336. except Exception as e:
  337. if retry_count >= 4 and minutes < 10:
  338. minutes += 1
  339. retry_count += 1
  340. s = "s" if minutes != 1 else ""
  341. self.log.exception(
  342. f"Error while syncing for refresh, retrying in {minutes} minute{s}"
  343. )
  344. if isinstance(e, IGCheckpointError):
  345. self.log.debug("Checkpoint error content: %s", e.body)
  346. await self.push_bridge_state(
  347. BridgeStateEvent.UNKNOWN_ERROR,
  348. error="unknown-error",
  349. message="An unknown error occurred while connecting to Instagram",
  350. info={"python_error": str(e)},
  351. )
  352. await asyncio.sleep(minutes * 60)
  353. else:
  354. self.start_listen()
  355. finally:
  356. self._is_refreshing = False
  357. async def _handle_checkpoint(
  358. self,
  359. e: IGChallengeError | IGConsentRequiredError,
  360. on: str,
  361. client: AndroidAPI | None = None,
  362. ) -> None:
  363. self.log.warning(f"Got checkpoint error on {on}: {e.body.serialize()}")
  364. client = client or self.client
  365. self.client = None
  366. self.mqtt = None
  367. if isinstance(e, IGConsentRequiredError):
  368. await self.push_bridge_state(
  369. BridgeStateEvent.BAD_CREDENTIALS,
  370. error="ig-consent-required",
  371. info=e.body.serialize(),
  372. )
  373. return
  374. error_code = "ig-checkpoint"
  375. try:
  376. resp = await client.challenge_reset()
  377. info = {
  378. "challenge_context": (
  379. resp.challenge_context.serialize() if resp.challenge_context_str else None
  380. ),
  381. "step_name": resp.step_name,
  382. "step_data": resp.step_data.serialize() if resp.step_data else None,
  383. "user_id": resp.user_id,
  384. "action": resp.action,
  385. "status": resp.status,
  386. "challenge": e.body.challenge.serialize() if e.body.challenge else None,
  387. }
  388. self.log.debug(f"Challenge state: {resp.serialize()}")
  389. if resp.challenge_context.challenge_type_enum == "HACKED_LOCK":
  390. error_code = "ig-checkpoint-locked"
  391. except Exception:
  392. self.log.exception("Error resetting challenge state")
  393. info = {"challenge": e.body.challenge.serialize() if e.body.challenge else None}
  394. await self.push_bridge_state(BridgeStateEvent.BAD_CREDENTIALS, error=error_code, info=info)
  395. async def _sync_thread(self, thread: Thread, allow_create: bool) -> None:
  396. portal = await po.Portal.get_by_thread(thread, self.igpk)
  397. if portal.mxid:
  398. self.log.debug(f"{thread.thread_id} has a portal, syncing and backfilling...")
  399. await portal.update_matrix_room(self, thread, backfill=True)
  400. elif allow_create:
  401. self.log.debug(f"{thread.thread_id} has been active recently, creating portal...")
  402. await portal.create_matrix_room(self, thread)
  403. else:
  404. self.log.debug(f"{thread.thread_id} is not active and doesn't have a portal")
  405. async def sync(self) -> None:
  406. sleep_minutes = 2
  407. while True:
  408. try:
  409. resp = await self.client.get_inbox()
  410. break
  411. except IGNotLoggedInError as e:
  412. self.log.exception("Got not logged in error while syncing")
  413. await self.logout(error=e)
  414. return
  415. except IGRateLimitError as e:
  416. self.log.error(
  417. "Got ratelimit error while trying to get inbox (%s), retrying in %d minutes",
  418. e.body,
  419. sleep_minutes,
  420. )
  421. await self.push_bridge_state(
  422. BridgeStateEvent.TRANSIENT_DISCONNECT, error="ig-rate-limit"
  423. )
  424. await asyncio.sleep(sleep_minutes * 60)
  425. sleep_minutes += 2
  426. except (IGChallengeError, IGConsentRequiredError) as e:
  427. await self._handle_checkpoint(e, on="sync")
  428. return
  429. self.seq_id = resp.seq_id
  430. self.snapshot_at_ms = resp.snapshot_at_ms
  431. await self.save_seq_id()
  432. if not self._listen_task:
  433. self.start_listen()
  434. max_age = self.config["bridge.portal_create_max_age"] * 1_000_000
  435. limit = self.config["bridge.chat_sync_limit"]
  436. create_limit = self.config["bridge.chat_create_limit"]
  437. min_active_at = (time.time() * 1_000_000) - max_age
  438. i = 0
  439. await self.push_bridge_state(BridgeStateEvent.BACKFILLING)
  440. async for thread in self.client.iter_inbox(start_at=resp):
  441. try:
  442. await self._sync_thread(
  443. thread=thread,
  444. allow_create=thread.last_activity_at > min_active_at and i < create_limit,
  445. )
  446. except Exception:
  447. self.log.exception(f"Error syncing thread {thread.thread_id}")
  448. i += 1
  449. if i >= limit:
  450. break
  451. try:
  452. await self.update_direct_chats()
  453. except Exception:
  454. self.log.exception("Error updating direct chat list")
  455. def start_listen(self) -> None:
  456. self.shutdown = False
  457. task = self._listen(seq_id=self.seq_id, snapshot_at_ms=self.snapshot_at_ms)
  458. self._listen_task = self.loop.create_task(task)
  459. async def _listen(self, seq_id: int, snapshot_at_ms: int) -> None:
  460. try:
  461. await self.mqtt.listen(
  462. graphql_subs={
  463. GraphQLSubscription.app_presence(),
  464. GraphQLSubscription.direct_typing(self.state.user_id),
  465. GraphQLSubscription.direct_status(),
  466. },
  467. skywalker_subs={
  468. SkywalkerSubscription.direct_sub(self.state.user_id),
  469. SkywalkerSubscription.live_sub(self.state.user_id),
  470. },
  471. seq_id=seq_id,
  472. snapshot_at_ms=snapshot_at_ms,
  473. )
  474. except IrisSubscribeError as e:
  475. self.log.warning(f"Got IrisSubscribeError {e}, refreshing...")
  476. asyncio.create_task(self.refresh())
  477. except (MQTTNotConnected, MQTTNotLoggedIn) as e:
  478. await self.send_bridge_notice(
  479. f"Error in listener: {e}",
  480. important=True,
  481. state_event=BridgeStateEvent.UNKNOWN_ERROR,
  482. error_code="ig-connection-error",
  483. )
  484. self.mqtt.disconnect()
  485. except Exception as e:
  486. self.log.exception("Fatal error in listener")
  487. await self.send_bridge_notice(
  488. "Fatal error in listener (see logs for more info)",
  489. state_event=BridgeStateEvent.UNKNOWN_ERROR,
  490. important=True,
  491. error_code="ig-connection-error",
  492. info={"python_error": str(e)},
  493. )
  494. self.mqtt.disconnect()
  495. else:
  496. if not self.shutdown:
  497. await self.send_bridge_notice(
  498. "Instagram connection closed without error",
  499. state_event=BridgeStateEvent.UNKNOWN_ERROR,
  500. error_code="ig-disconnected",
  501. )
  502. finally:
  503. self._listen_task = None
  504. self._is_connected = False
  505. self._track_metric(METRIC_CONNECTED, False)
  506. async def stop_listen(self) -> None:
  507. if self.mqtt:
  508. self.shutdown = True
  509. self.mqtt.disconnect()
  510. if self._listen_task:
  511. await self._listen_task
  512. self.shutdown = False
  513. self._track_metric(METRIC_CONNECTED, False)
  514. self._is_connected = False
  515. await self.update()
  516. async def logout(self, error: IGNotLoggedInError | None = None) -> None:
  517. if self.client and error is None:
  518. try:
  519. await self.client.logout(one_tap_app_login=False)
  520. except Exception:
  521. self.log.debug("Exception logging out", exc_info=True)
  522. if self.mqtt:
  523. self.mqtt.disconnect()
  524. self._track_metric(METRIC_CONNECTED, False)
  525. self._track_metric(METRIC_LOGGED_IN, False)
  526. if error is None:
  527. await self.push_bridge_state(BridgeStateEvent.LOGGED_OUT)
  528. puppet = await pu.Puppet.get_by_pk(self.igpk, create=False)
  529. if puppet and puppet.is_real_user:
  530. await puppet.switch_mxid(None, None)
  531. try:
  532. del self.by_igpk[self.igpk]
  533. except KeyError:
  534. pass
  535. self.igpk = None
  536. else:
  537. self.log.debug("Auth error body: %s", error.body.serialize())
  538. await self.send_bridge_notice(
  539. f"You have been logged out of Instagram: {error.proper_message}",
  540. important=True,
  541. state_event=BridgeStateEvent.BAD_CREDENTIALS,
  542. error_code="ig-auth-error",
  543. error_message=error.proper_message,
  544. )
  545. self.client = None
  546. self.mqtt = None
  547. self.state = None
  548. self.seq_id = None
  549. self.snapshot_at_ms = None
  550. self._is_logged_in = False
  551. await self.update()
  552. # endregion
  553. # region Event handlers
  554. async def _save_seq_id_after_sleep(self) -> None:
  555. await asyncio.sleep(120)
  556. self._seq_id_save_task = None
  557. self.log.trace("Saving sequence ID %d/%d", self.seq_id, self.snapshot_at_ms)
  558. try:
  559. await self.save_seq_id()
  560. except Exception:
  561. self.log.exception("Error saving sequence ID")
  562. async def update_seq_id(self, evt: NewSequenceID) -> None:
  563. self.seq_id = evt.seq_id
  564. self.snapshot_at_ms = evt.snapshot_at_ms
  565. if not self._seq_id_save_task or self._seq_id_save_task.done():
  566. self.log.trace("Starting seq id save task (%d/%d)", evt.seq_id, evt.snapshot_at_ms)
  567. self._seq_id_save_task = asyncio.create_task(self._save_seq_id_after_sleep())
  568. else:
  569. self.log.trace("Not starting seq id save task (%d/%d)", evt.seq_id, evt.snapshot_at_ms)
  570. @async_time(METRIC_MESSAGE)
  571. async def handle_message(self, evt: MessageSyncEvent) -> None:
  572. portal = await po.Portal.get_by_thread_id(evt.message.thread_id, receiver=self.igpk)
  573. if not portal or not portal.mxid:
  574. self.log.debug("Got message in thread with no portal, getting info...")
  575. resp = await self.client.get_thread(evt.message.thread_id)
  576. portal = await po.Portal.get_by_thread(resp.thread, self.igpk)
  577. self.log.debug("Got info for unknown portal, creating room")
  578. await portal.create_matrix_room(self, resp.thread)
  579. if not portal.mxid:
  580. self.log.warning(
  581. "Room creation appears to have failed, "
  582. f"dropping message in {evt.message.thread_id}"
  583. )
  584. return
  585. self.log.trace(f"Received message sync event {evt.message}")
  586. sender = await pu.Puppet.get_by_pk(evt.message.user_id) if evt.message.user_id else None
  587. if evt.message.op == Operation.ADD:
  588. if not sender:
  589. # I don't think we care about adds with no sender
  590. return
  591. await portal.handle_instagram_item(self, sender, evt.message)
  592. elif evt.message.op == Operation.REMOVE:
  593. # Removes don't have a sender, only the message sender can unsend messages anyway
  594. await portal.handle_instagram_remove(evt.message.item_id)
  595. elif evt.message.op == Operation.REPLACE:
  596. await portal.handle_instagram_update(evt.message)
  597. @async_time(METRIC_THREAD_SYNC)
  598. async def handle_thread_sync(self, evt: ThreadSyncEvent) -> None:
  599. self.log.trace("Received thread sync event %s", evt)
  600. portal = await po.Portal.get_by_thread(evt, receiver=self.igpk)
  601. await portal.create_matrix_room(self, evt)
  602. @async_time(METRIC_RTD)
  603. async def handle_rtd(self, evt: RealtimeDirectEvent) -> None:
  604. if not isinstance(evt.value, ActivityIndicatorData):
  605. return
  606. now = int(time.time() * 1000)
  607. date = evt.value.timestamp_ms
  608. expiry = date + evt.value.ttl
  609. if expiry < now:
  610. return
  611. if evt.activity_indicator_id in self._activity_indicator_ids:
  612. return
  613. # TODO clear expired items from this dict
  614. self._activity_indicator_ids[evt.activity_indicator_id] = expiry
  615. puppet = await pu.Puppet.get_by_pk(int(evt.value.sender_id))
  616. portal = await po.Portal.get_by_thread_id(evt.thread_id, receiver=self.igpk)
  617. if not puppet or not portal or not portal.mxid:
  618. return
  619. is_typing = evt.value.activity_status != TypingStatus.OFF
  620. if puppet.pk == self.igpk:
  621. self.remote_typing_status = TypingStatus.TEXT if is_typing else TypingStatus.OFF
  622. await puppet.intent_for(portal).set_typing(
  623. portal.mxid, is_typing=is_typing, timeout=evt.value.ttl
  624. )
  625. # endregion
  626. # region Database getters
  627. def _add_to_cache(self) -> None:
  628. self.by_mxid[self.mxid] = self
  629. if self.igpk:
  630. self.by_igpk[self.igpk] = self
  631. @classmethod
  632. @async_getter_lock
  633. async def get_by_mxid(cls, mxid: UserID, *, create: bool = True) -> User | None:
  634. # Never allow ghosts to be users
  635. if pu.Puppet.get_id_from_mxid(mxid):
  636. return None
  637. try:
  638. return cls.by_mxid[mxid]
  639. except KeyError:
  640. pass
  641. user = cast(cls, await super().get_by_mxid(mxid))
  642. if user is not None:
  643. user._add_to_cache()
  644. return user
  645. if create:
  646. user = cls(mxid)
  647. await user.insert()
  648. user._add_to_cache()
  649. return user
  650. return None
  651. @classmethod
  652. @async_getter_lock
  653. async def get_by_igpk(cls, igpk: int) -> User | None:
  654. try:
  655. return cls.by_igpk[igpk]
  656. except KeyError:
  657. pass
  658. user = cast(cls, await super().get_by_igpk(igpk))
  659. if user is not None:
  660. user._add_to_cache()
  661. return user
  662. return None
  663. @classmethod
  664. async def all_logged_in(cls) -> AsyncGenerator[User, None]:
  665. users = await super().all_logged_in()
  666. user: cls
  667. for index, user in enumerate(users):
  668. try:
  669. yield cls.by_mxid[user.mxid]
  670. except KeyError:
  671. user._add_to_cache()
  672. yield user
  673. # endregion