user.py 47 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205
  1. # mautrix-instagram - A Matrix-Instagram puppeting bridge.
  2. # Copyright (C) 2023 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, Callable, cast
  18. from datetime import datetime, timedelta
  19. from functools import partial
  20. import asyncio
  21. import logging
  22. import time
  23. from aiohttp import ClientConnectionError
  24. from mauigpapi import AndroidAPI, AndroidMQTT, AndroidState, ProxyHandler
  25. from mauigpapi.errors import (
  26. IGChallengeError,
  27. IGCheckpointError,
  28. IGConsentRequiredError,
  29. IGNotLoggedInError,
  30. IGRateLimitError,
  31. IGUserIDNotFoundError,
  32. IrisSubscribeError,
  33. MQTTConnectionUnauthorized,
  34. MQTTNotConnected,
  35. MQTTNotLoggedIn,
  36. MQTTReconnectionError,
  37. )
  38. from mauigpapi.mqtt import (
  39. Connect,
  40. Disconnect,
  41. GraphQLSubscription,
  42. NewSequenceID,
  43. ProxyUpdate,
  44. SkywalkerSubscription,
  45. )
  46. from mauigpapi.types import (
  47. ActivityIndicatorData,
  48. CurrentUser,
  49. MessageSyncEvent,
  50. Operation,
  51. RealtimeDirectEvent,
  52. Thread,
  53. ThreadRemoveEvent,
  54. ThreadSyncEvent,
  55. TypingStatus,
  56. )
  57. from mauigpapi.types.direct_inbox import DMInbox, DMInboxResponse
  58. from mautrix.appservice import AppService
  59. from mautrix.bridge import BaseUser, async_getter_lock
  60. from mautrix.types import EventID, MessageType, RoomID, TextMessageEventContent, UserID
  61. from mautrix.util.bridge_state import BridgeState, BridgeStateEvent
  62. from mautrix.util.logging import TraceLogger
  63. from mautrix.util.opt_prometheus import Gauge, Summary, async_time
  64. from mautrix.util.simple_lock import SimpleLock
  65. from . import portal as po, puppet as pu
  66. from .config import Config
  67. from .db import Backfill, Message as DBMessage, Portal as DBPortal, User as DBUser
  68. if TYPE_CHECKING:
  69. from .__main__ import InstagramBridge
  70. try:
  71. from aiohttp_socks import ProxyConnectionError, ProxyError, ProxyTimeoutError
  72. except ImportError:
  73. class ProxyError(Exception):
  74. pass
  75. ProxyConnectionError = ProxyTimeoutError = ProxyError
  76. METRIC_MESSAGE = Summary("bridge_on_message", "calls to handle_message")
  77. METRIC_THREAD_SYNC = Summary("bridge_on_thread_sync", "calls to handle_thread_sync")
  78. METRIC_RTD = Summary("bridge_on_rtd", "calls to handle_rtd")
  79. METRIC_LOGGED_IN = Gauge("bridge_logged_in", "Users logged into the bridge")
  80. METRIC_CONNECTED = Gauge("bridge_connected", "Bridged users connected to Instagram")
  81. BridgeState.human_readable_errors.update(
  82. {
  83. "ig-connection-error": "Instagram disconnected unexpectedly",
  84. "ig-refresh-connection-error": "Reconnecting failed again after refresh: {message}",
  85. "ig-connection-fatal-error": "Instagram disconnected unexpectedly",
  86. "ig-auth-error": "Authentication error from Instagram: {message}",
  87. "ig-checkpoint": "Instagram checkpoint error. Please check the Instagram website.",
  88. "ig-consent-required": "Instagram requires a consent update. Please check the Instagram website.",
  89. "ig-checkpoint-locked": "Instagram checkpoint error. Please check the Instagram website.",
  90. "ig-rate-limit": "Got Instagram ratelimit error, waiting a few minutes before retrying...",
  91. "ig-disconnected": None,
  92. "ig-no-mqtt": "You're not connected to Instagram",
  93. "logged-out": "You're not logged into Instagram",
  94. }
  95. )
  96. class User(DBUser, BaseUser):
  97. ig_base_log: TraceLogger = logging.getLogger("mau.instagram")
  98. _activity_indicator_ids: dict[str, int] = {}
  99. by_mxid: dict[UserID, User] = {}
  100. by_igpk: dict[int, User] = {}
  101. config: Config
  102. az: AppService
  103. loop: asyncio.AbstractEventLoop
  104. client: AndroidAPI | None
  105. mqtt: AndroidMQTT | None
  106. _listen_task: asyncio.Task | None = None
  107. _sync_lock: SimpleLock
  108. _backfill_loop_task: asyncio.Task | None
  109. _thread_sync_task: asyncio.Task | None
  110. _seq_id_save_task: asyncio.Task | None
  111. permission_level: str
  112. username: str | None
  113. _notice_room_lock: asyncio.Lock
  114. _notice_send_lock: asyncio.Lock
  115. _is_logged_in: bool
  116. _is_connected: bool
  117. shutdown: bool
  118. remote_typing_status: TypingStatus | None
  119. def __init__(
  120. self,
  121. mxid: UserID,
  122. igpk: int | None = None,
  123. state: AndroidState | None = None,
  124. notice_room: RoomID | None = None,
  125. seq_id: int | None = None,
  126. snapshot_at_ms: int | None = None,
  127. oldest_cursor: str | None = None,
  128. total_backfilled_portals: int | None = None,
  129. thread_sync_completed: bool = False,
  130. ) -> None:
  131. super().__init__(
  132. mxid=mxid,
  133. igpk=igpk,
  134. state=state,
  135. notice_room=notice_room,
  136. seq_id=seq_id,
  137. snapshot_at_ms=snapshot_at_ms,
  138. oldest_cursor=oldest_cursor,
  139. total_backfilled_portals=total_backfilled_portals,
  140. thread_sync_completed=thread_sync_completed,
  141. )
  142. BaseUser.__init__(self)
  143. self._notice_room_lock = asyncio.Lock()
  144. self._notice_send_lock = asyncio.Lock()
  145. perms = self.config.get_permissions(mxid)
  146. self.relay_whitelisted, self.is_whitelisted, self.is_admin, self.permission_level = perms
  147. self.client = None
  148. self.mqtt = None
  149. self.username = None
  150. self._is_logged_in = False
  151. self._is_connected = False
  152. self._is_refreshing = False
  153. self.shutdown = False
  154. self._sync_lock = SimpleLock(
  155. "Waiting for thread sync to finish before handling %s", log=self.log
  156. )
  157. self._listen_task = None
  158. self._thread_sync_task = None
  159. self._backfill_loop_task = None
  160. self.remote_typing_status = None
  161. self._seq_id_save_task = None
  162. self.proxy_handler = ProxyHandler(
  163. api_url=self.config["bridge.get_proxy_api_url"],
  164. )
  165. @classmethod
  166. def init_cls(cls, bridge: "InstagramBridge") -> AsyncIterable[Awaitable[None]]:
  167. cls.bridge = bridge
  168. cls.config = bridge.config
  169. cls.az = bridge.az
  170. cls.loop = bridge.loop
  171. return (user.try_connect() async for user in cls.all_logged_in())
  172. # region Connection management
  173. async def is_logged_in(self) -> bool:
  174. return bool(self.client) and self._is_logged_in
  175. async def get_puppet(self) -> pu.Puppet | None:
  176. if not self.igpk:
  177. return None
  178. return await pu.Puppet.get_by_pk(self.igpk)
  179. async def get_portal_with(self, puppet: pu.Puppet, create: bool = True) -> po.Portal | None:
  180. if not self.igpk:
  181. return None
  182. portal = await po.Portal.find_private_chat(self.igpk, puppet.pk)
  183. if portal:
  184. return portal
  185. if create:
  186. # TODO add error handling somewhere
  187. thread = await self.client.create_group_thread([puppet.pk])
  188. portal = await po.Portal.get_by_thread(thread, self.igpk)
  189. await portal.update_info(thread, self)
  190. return portal
  191. return None
  192. async def try_connect(self) -> None:
  193. try:
  194. await self.connect()
  195. except Exception as e:
  196. self.log.exception("Error while connecting to Instagram")
  197. await self.push_bridge_state(
  198. BridgeStateEvent.UNKNOWN_ERROR, info={"python_error": str(e)}
  199. )
  200. @property
  201. def api_log(self) -> TraceLogger:
  202. return self.ig_base_log.getChild("http").getChild(self.mxid)
  203. @property
  204. def is_connected(self) -> bool:
  205. return bool(self.client) and bool(self.mqtt) and self._is_connected
  206. async def connect(self, user: CurrentUser | None = None) -> None:
  207. if not self.state:
  208. await self.push_bridge_state(BridgeStateEvent.BAD_CREDENTIALS, error="logged-out")
  209. return
  210. client = AndroidAPI(
  211. self.state,
  212. log=self.api_log,
  213. proxy_handler=self.proxy_handler,
  214. )
  215. if not user:
  216. try:
  217. resp = await client.current_user()
  218. user = resp.user
  219. except IGNotLoggedInError as e:
  220. self.log.warning(f"Failed to connect to Instagram: {e}, logging out")
  221. await self.logout(error=e)
  222. return
  223. except IGCheckpointError as e:
  224. self.log.debug("Checkpoint error content: %s", e.body)
  225. raise
  226. except (IGChallengeError, IGConsentRequiredError) as e:
  227. await self._handle_checkpoint(e, on="connect", client=client)
  228. return
  229. self.client = client
  230. self._is_logged_in = True
  231. self.igpk = user.pk
  232. self.username = user.username
  233. await self.push_bridge_state(BridgeStateEvent.CONNECTING)
  234. self._track_metric(METRIC_LOGGED_IN, True)
  235. self.by_igpk[self.igpk] = self
  236. self.mqtt = AndroidMQTT(
  237. self.state,
  238. log=self.ig_base_log.getChild("mqtt").getChild(self.mxid),
  239. proxy_handler=self.proxy_handler,
  240. )
  241. self.mqtt.add_event_handler(Connect, self.on_connect)
  242. self.mqtt.add_event_handler(Disconnect, self.on_disconnect)
  243. self.mqtt.add_event_handler(NewSequenceID, self.update_seq_id)
  244. self.mqtt.add_event_handler(MessageSyncEvent, self.handle_message)
  245. self.mqtt.add_event_handler(ThreadSyncEvent, self.handle_thread_sync)
  246. self.mqtt.add_event_handler(ThreadRemoveEvent, self.handle_thread_remove)
  247. self.mqtt.add_event_handler(RealtimeDirectEvent, self.handle_rtd)
  248. self.mqtt.add_event_handler(ProxyUpdate, self.on_proxy_update)
  249. await self.update()
  250. self.loop.create_task(self._try_sync_puppet(user))
  251. self.loop.create_task(self._post_connect())
  252. async def _post_connect(self):
  253. # Backfill requests are handled synchronously so as not to overload the homeserver.
  254. # Users can configure their backfill stages to be more or less aggressive with backfilling
  255. # to try and avoid getting banned.
  256. if not self._backfill_loop_task or self._backfill_loop_task.done():
  257. self._backfill_loop_task = asyncio.create_task(self._handle_backfill_requests_loop())
  258. if not self.seq_id:
  259. await self._try_sync()
  260. else:
  261. self.log.debug("Connecting to MQTT directly as resync_on_startup is false")
  262. self.start_listen()
  263. if self.config["bridge.backfill.enable"]:
  264. if self._thread_sync_task and not self._thread_sync_task.done():
  265. self.log.warning("Cancelling existing background thread sync task")
  266. self._thread_sync_task.cancel()
  267. self._thread_sync_task = asyncio.create_task(self.backfill_threads())
  268. async def _handle_backfill_requests_loop(self) -> None:
  269. if not self.config["bridge.backfill.enable"] or not self.config["bridge.backfill.msc2716"]:
  270. return
  271. while True:
  272. await self._sync_lock.wait("backfill request")
  273. req = await Backfill.get_next(self.mxid)
  274. if not req:
  275. await asyncio.sleep(30)
  276. continue
  277. self.log.info("Backfill request %s", req)
  278. try:
  279. portal = await po.Portal.get_by_thread_id(
  280. req.portal_thread_id, receiver=req.portal_receiver
  281. )
  282. await req.mark_dispatched()
  283. await portal.backfill(self, req)
  284. await req.mark_done()
  285. except IGNotLoggedInError as e:
  286. self.log.exception("User got logged out during backfill loop")
  287. await self.logout(error=e)
  288. break
  289. except (IGChallengeError, IGConsentRequiredError) as e:
  290. self.log.exception("User got a challenge during backfill loop")
  291. await self._handle_checkpoint(e, on="backfill")
  292. break
  293. except Exception as e:
  294. self.log.exception("Failed to backfill portal %s: %s", req.portal_thread_id, e)
  295. # Don't try again to backfill this portal for a minute.
  296. await req.set_cooldown_timeout(60)
  297. self._backfill_loop_task = None
  298. async def on_connect(self, evt: Connect) -> None:
  299. self.log.debug("Connected to Instagram")
  300. self._track_metric(METRIC_CONNECTED, True)
  301. self._is_connected = True
  302. await self.send_bridge_notice("Connected to Instagram")
  303. await self.push_bridge_state(BridgeStateEvent.CONNECTED)
  304. async def on_disconnect(self, evt: Disconnect) -> None:
  305. self.log.debug("Disconnected from Instagram")
  306. self._track_metric(METRIC_CONNECTED, False)
  307. self._is_connected = False
  308. async def on_proxy_update(self, evt: ProxyUpdate | None = None) -> None:
  309. if self.client:
  310. self.client.setup_http(self.state.cookies.jar)
  311. # TODO this stuff could probably be moved to mautrix-python
  312. async def get_notice_room(self) -> RoomID:
  313. if not self.notice_room:
  314. async with self._notice_room_lock:
  315. # If someone already created the room while this call was waiting,
  316. # don't make a new room
  317. if self.notice_room:
  318. return self.notice_room
  319. creation_content = {}
  320. if not self.config["bridge.federate_rooms"]:
  321. creation_content["m.federate"] = False
  322. self.notice_room = await self.az.intent.create_room(
  323. is_direct=True,
  324. invitees=[self.mxid],
  325. topic="Instagram bridge notices",
  326. creation_content=creation_content,
  327. )
  328. await self.update()
  329. return self.notice_room
  330. async def fill_bridge_state(self, state: BridgeState) -> None:
  331. await super().fill_bridge_state(state)
  332. if not state.remote_id:
  333. if self.igpk:
  334. state.remote_id = str(self.igpk)
  335. else:
  336. try:
  337. state.remote_id = self.state.user_id
  338. except IGUserIDNotFoundError:
  339. state.remote_id = None
  340. if self.username:
  341. state.remote_name = f"@{self.username}"
  342. async def get_bridge_states(self) -> list[BridgeState]:
  343. if not self.state:
  344. return []
  345. state = BridgeState(state_event=BridgeStateEvent.UNKNOWN_ERROR)
  346. if self.is_connected:
  347. state.state_event = BridgeStateEvent.CONNECTED
  348. elif self._is_refreshing or self.mqtt:
  349. state.state_event = BridgeStateEvent.TRANSIENT_DISCONNECT
  350. return [state]
  351. async def send_bridge_notice(
  352. self,
  353. text: str,
  354. edit: EventID | None = None,
  355. state_event: BridgeStateEvent | None = None,
  356. important: bool = False,
  357. error_code: str | None = None,
  358. error_message: str | None = None,
  359. info: dict | None = None,
  360. ) -> EventID | None:
  361. if state_event:
  362. await self.push_bridge_state(
  363. state_event,
  364. error=error_code,
  365. message=error_message if error_code else text,
  366. info=info,
  367. )
  368. if self.config["bridge.disable_bridge_notices"]:
  369. return None
  370. if not important and not self.config["bridge.unimportant_bridge_notices"]:
  371. self.log.debug("Not sending unimportant bridge notice: %s", text)
  372. return None
  373. event_id = None
  374. try:
  375. self.log.debug("Sending bridge notice: %s", text)
  376. content = TextMessageEventContent(
  377. body=text, msgtype=(MessageType.TEXT if important else MessageType.NOTICE)
  378. )
  379. if edit:
  380. content.set_edit(edit)
  381. # This is locked to prevent notices going out in the wrong order
  382. async with self._notice_send_lock:
  383. event_id = await self.az.intent.send_message(await self.get_notice_room(), content)
  384. except Exception:
  385. self.log.warning("Failed to send bridge notice", exc_info=True)
  386. return edit or event_id
  387. async def _try_sync_puppet(self, user_info: CurrentUser) -> None:
  388. puppet = await pu.Puppet.get_by_pk(self.igpk)
  389. try:
  390. await puppet.update_info(user_info, self)
  391. except Exception:
  392. self.log.exception("Failed to update own puppet info")
  393. try:
  394. if puppet.custom_mxid != self.mxid and puppet.can_auto_login(self.mxid):
  395. self.log.info(f"Automatically enabling custom puppet")
  396. await puppet.switch_mxid(access_token="auto", mxid=self.mxid)
  397. except Exception:
  398. self.log.exception("Failed to automatically enable custom puppet")
  399. async def _try_sync(self) -> None:
  400. try:
  401. await self.sync()
  402. except Exception as e:
  403. self.log.exception("Exception while syncing")
  404. if isinstance(e, IGCheckpointError):
  405. self.log.debug("Checkpoint error content: %s", e.body)
  406. await self.push_bridge_state(
  407. BridgeStateEvent.UNKNOWN_ERROR, info={"python_error": str(e)}
  408. )
  409. async def get_direct_chats(self) -> dict[UserID, list[RoomID]]:
  410. return {
  411. pu.Puppet.get_mxid_from_id(portal.other_user_pk): [portal.mxid]
  412. for portal in await DBPortal.find_private_chats_of(self.igpk)
  413. if portal.mxid
  414. }
  415. async def refresh(self, resync: bool = True) -> None:
  416. self._is_refreshing = True
  417. try:
  418. await self.stop_listen()
  419. self.state.reset_pigeon_session_id()
  420. if resync:
  421. retry_count = 0
  422. minutes = 1
  423. while True:
  424. try:
  425. await self.sync()
  426. return
  427. except Exception as e:
  428. if retry_count >= 4 and minutes < 10:
  429. minutes += 1
  430. retry_count += 1
  431. s = "s" if minutes != 1 else ""
  432. self.log.exception(
  433. f"Error while syncing for refresh, retrying in {minutes} minute{s}"
  434. )
  435. if isinstance(e, IGCheckpointError):
  436. self.log.debug("Checkpoint error content: %s", e.body)
  437. await self.push_bridge_state(
  438. BridgeStateEvent.UNKNOWN_ERROR,
  439. error="unknown-error",
  440. message="An unknown error occurred while connecting to Instagram",
  441. info={"python_error": str(e)},
  442. )
  443. await asyncio.sleep(minutes * 60)
  444. else:
  445. self.start_listen()
  446. finally:
  447. self._is_refreshing = False
  448. self.proxy_handler.update_proxy_url()
  449. async def _handle_checkpoint(
  450. self,
  451. e: IGChallengeError | IGConsentRequiredError,
  452. on: str,
  453. client: AndroidAPI | None = None,
  454. ) -> None:
  455. self.log.warning(f"Got checkpoint error on {on}: {e.body.serialize()}")
  456. client = client or self.client
  457. self.client = None
  458. self.mqtt = None
  459. if isinstance(e, IGConsentRequiredError):
  460. await self.push_bridge_state(
  461. BridgeStateEvent.BAD_CREDENTIALS,
  462. error="ig-consent-required",
  463. info=e.body.serialize(),
  464. )
  465. return
  466. error_code = "ig-checkpoint"
  467. try:
  468. resp = await client.challenge_reset()
  469. info = {
  470. "challenge_context": (
  471. resp.challenge_context.serialize() if resp.challenge_context_str else None
  472. ),
  473. "step_name": resp.step_name,
  474. "step_data": resp.step_data.serialize() if resp.step_data else None,
  475. "user_id": resp.user_id,
  476. "action": resp.action,
  477. "status": resp.status,
  478. "challenge": e.body.challenge.serialize() if e.body.challenge else None,
  479. }
  480. self.log.debug(f"Challenge state: {resp.serialize()}")
  481. if resp.challenge_context.challenge_type_enum == "HACKED_LOCK":
  482. error_code = "ig-checkpoint-locked"
  483. except Exception:
  484. self.log.exception("Error resetting challenge state")
  485. info = {"challenge": e.body.challenge.serialize() if e.body.challenge else None}
  486. await self.push_bridge_state(BridgeStateEvent.BAD_CREDENTIALS, error=error_code, info=info)
  487. async def _sync_thread(self, thread: Thread) -> bool:
  488. """
  489. Sync a specific thread. Returns whether the thread had messages after the last message in
  490. the database before the sync.
  491. """
  492. self.log.debug(f"Syncing thread {thread.thread_id}")
  493. forward_messages = thread.items
  494. assert self.client
  495. portal = await po.Portal.get_by_thread(thread, self.igpk)
  496. assert portal
  497. # Create or update the Matrix room
  498. if not portal.mxid:
  499. await portal.create_matrix_room(self, thread)
  500. else:
  501. await portal.update_matrix_room(self, thread)
  502. if not self.config["bridge.backfill.enable_initial"]:
  503. return True
  504. last_message = await DBMessage.get_last(portal.mxid)
  505. cursor = thread.oldest_cursor
  506. if last_message:
  507. original_number_of_messages = len(thread.items)
  508. new_messages = [
  509. m for m in thread.items if last_message.ig_timestamp_ms < m.timestamp_ms
  510. ]
  511. forward_messages = new_messages
  512. portal.log.debug(
  513. f"{len(new_messages)}/{original_number_of_messages} messages are after most recent"
  514. " message."
  515. )
  516. # Fetch more messages until we get back to messages that have been bridged already.
  517. while len(new_messages) > 0 and len(new_messages) == original_number_of_messages:
  518. await asyncio.sleep(self.config["bridge.backfill.incremental.page_delay"])
  519. portal.log.debug("Fetching more messages for forward backfill")
  520. resp = await self.client.get_thread(portal.thread_id, cursor=cursor)
  521. if len(resp.thread.items) == 0:
  522. break
  523. original_number_of_messages = len(resp.thread.items)
  524. new_messages = [
  525. m for m in resp.thread.items if last_message.ig_timestamp_ms < m.timestamp_ms
  526. ]
  527. forward_messages = new_messages + forward_messages
  528. cursor = resp.thread.oldest_cursor
  529. portal.log.debug(
  530. f"{len(new_messages)}/{original_number_of_messages} messages are after most "
  531. "recent message."
  532. )
  533. elif not portal.first_event_id:
  534. self.log.debug(
  535. f"Skipping backfilling {portal.thread_id} as the first event ID is not known"
  536. )
  537. return False
  538. if forward_messages:
  539. portal.cursor = cursor
  540. await portal.update()
  541. mark_read = thread.read_state == 0 or (
  542. (hours := self.config["bridge.backfill.unread_hours_threshold"]) > 0
  543. and (
  544. datetime.fromtimestamp(forward_messages[0].timestamp_ms / 1000)
  545. < datetime.now() - timedelta(hours=hours)
  546. )
  547. )
  548. base_insertion_event_id = await portal.backfill_message_page(
  549. self,
  550. list(reversed(forward_messages)),
  551. forward=True,
  552. last_message=last_message,
  553. mark_read=mark_read,
  554. )
  555. if (
  556. not self.bridge.homeserver_software.is_hungry
  557. and self.config["bridge.backfill.msc2716"]
  558. ):
  559. await portal.send_post_backfill_dummy(
  560. forward_messages[0].timestamp, base_insertion_event_id=base_insertion_event_id
  561. )
  562. if (
  563. mark_read
  564. and not self.bridge.homeserver_software.is_hungry
  565. and (puppet := await self.get_puppet())
  566. ):
  567. last_message = await DBMessage.get_last(portal.mxid)
  568. if last_message:
  569. await puppet.intent_for(portal).mark_read(portal.mxid, last_message.mxid)
  570. await portal._update_read_receipts(thread.last_seen_at)
  571. if self.config["bridge.backfill.msc2716"]:
  572. await portal.enqueue_immediate_backfill(self, 1)
  573. return len(forward_messages) > 0
  574. async def _maybe_update_proxy(self, source: str) -> None:
  575. if not self._listen_task:
  576. self.proxy_handler.update_proxy_url()
  577. await self.on_proxy_update()
  578. else:
  579. self.log.debug(f"Not updating proxy: listen_task is still running? (caller: {source})")
  580. async def sync(self, increment_total_backfilled_portals: bool = False) -> None:
  581. await self.run_with_sync_lock(partial(self._sync, increment_total_backfilled_portals))
  582. async def _sync(self, increment_total_backfilled_portals: bool = False) -> None:
  583. if not self._listen_task:
  584. self.state.reset_pigeon_session_id()
  585. sleep_minutes = 2
  586. errors = 0
  587. while True:
  588. try:
  589. resp = await self.client.get_inbox()
  590. break
  591. except (
  592. ProxyError,
  593. ProxyTimeoutError,
  594. ProxyConnectionError,
  595. ClientConnectionError,
  596. ConnectionError,
  597. asyncio.TimeoutError,
  598. ) as e:
  599. errors += 1
  600. wait = min(errors * 10, 60)
  601. self.log.warning(
  602. f"{e.__class__.__name__} while trying to sync, retrying in {wait} seconds: {e}"
  603. )
  604. await asyncio.sleep(wait)
  605. await self._maybe_update_proxy("sync error")
  606. except IGNotLoggedInError as e:
  607. self.log.exception("Got not logged in error while syncing")
  608. await self.logout(error=e)
  609. return
  610. except IGRateLimitError as e:
  611. self.log.error(
  612. "Got ratelimit error while trying to get inbox (%s), retrying in %d minutes",
  613. e.body,
  614. sleep_minutes,
  615. )
  616. await self.push_bridge_state(
  617. BridgeStateEvent.TRANSIENT_DISCONNECT, error="ig-rate-limit"
  618. )
  619. await asyncio.sleep(sleep_minutes * 60)
  620. sleep_minutes += 2
  621. except IGCheckpointError as e:
  622. self.log.debug("Checkpoint error content: %s", e.body)
  623. raise
  624. except (IGChallengeError, IGConsentRequiredError) as e:
  625. await self._handle_checkpoint(e, on="sync")
  626. return
  627. self.seq_id = resp.seq_id
  628. self.snapshot_at_ms = resp.snapshot_at_ms
  629. await self.save_seq_id()
  630. if not self._listen_task:
  631. self.start_listen(is_after_sync=True)
  632. sync_count = min(
  633. self.config["bridge.backfill.max_conversations"],
  634. self.config["bridge.max_startup_thread_sync_count"],
  635. )
  636. self.log.debug(f"Fetching {sync_count} threads, 20 at a time...")
  637. local_limit: int | None = sync_count
  638. if sync_count == 0:
  639. return
  640. elif sync_count < 0:
  641. local_limit = None
  642. await self._sync_threads_with_delay(
  643. self.client.iter_inbox(
  644. self._update_seq_id_and_cursor, start_at=resp, local_limit=local_limit
  645. ),
  646. stop_when_threads_have_no_messages_to_backfill=True,
  647. increment_total_backfilled_portals=increment_total_backfilled_portals,
  648. local_limit=local_limit,
  649. )
  650. try:
  651. await self.update_direct_chats()
  652. except Exception:
  653. self.log.exception("Error updating direct chat list")
  654. async def backfill_threads(self):
  655. try:
  656. await self.run_with_sync_lock(self._backfill_threads)
  657. except Exception:
  658. self.log.exception("Error in thread backfill loop")
  659. async def _backfill_threads(self):
  660. assert self.client
  661. if not self.config["bridge.backfill.enable"]:
  662. return
  663. max_conversations = self.config["bridge.backfill.max_conversations"] or 0
  664. if 0 <= max_conversations <= (self.total_backfilled_portals or 0):
  665. self.log.info("Backfill max_conversations count reached, not syncing any more portals")
  666. return
  667. elif self.thread_sync_completed:
  668. self.log.debug("Thread backfill is marked as completed, not syncing more portals")
  669. return
  670. local_limit = (
  671. max_conversations - (self.total_backfilled_portals or 0)
  672. if max_conversations >= 0
  673. else None
  674. )
  675. start_at = None
  676. if self.oldest_cursor:
  677. start_at = DMInboxResponse(
  678. status="",
  679. seq_id=self.seq_id,
  680. snapshot_at_ms=0,
  681. pending_requests_total=0,
  682. has_pending_top_requests=False,
  683. viewer=None,
  684. inbox=DMInbox(
  685. threads=[],
  686. has_older=True,
  687. unseen_count=0,
  688. unseen_count_ts=0,
  689. blended_inbox_enabled=False,
  690. oldest_cursor=self.oldest_cursor,
  691. ),
  692. )
  693. backoff = self.config.get("bridge.backfill.backoff.thread_list", 300)
  694. await self._sync_threads_with_delay(
  695. self.client.iter_inbox(
  696. self._update_seq_id_and_cursor,
  697. start_at=start_at,
  698. local_limit=local_limit,
  699. rate_limit_exceeded_backoff=backoff,
  700. ),
  701. increment_total_backfilled_portals=True,
  702. local_limit=local_limit,
  703. )
  704. await self.update_direct_chats()
  705. def _update_seq_id_and_cursor(self, seq_id: int, cursor: str | None):
  706. self.seq_id = seq_id
  707. if cursor:
  708. self.oldest_cursor = cursor
  709. async def _sync_threads_with_delay(
  710. self,
  711. threads: AsyncIterable[Thread],
  712. increment_total_backfilled_portals: bool = False,
  713. stop_when_threads_have_no_messages_to_backfill: bool = False,
  714. local_limit: int | None = None,
  715. ):
  716. sync_delay = self.config["bridge.backfill.min_sync_thread_delay"]
  717. last_thread_sync_ts = 0.0
  718. found_thread_count = 0
  719. async for thread in threads:
  720. found_thread_count += 1
  721. now = time.monotonic()
  722. if now < last_thread_sync_ts + sync_delay:
  723. delay = last_thread_sync_ts + sync_delay - now
  724. self.log.debug("Thread sync is happening too quickly. Waiting for %ds", delay)
  725. await asyncio.sleep(delay)
  726. last_thread_sync_ts = time.monotonic()
  727. had_new_messages = await self._sync_thread(thread)
  728. if not had_new_messages and stop_when_threads_have_no_messages_to_backfill:
  729. self.log.debug("Got to threads with no new messages. Stopping sync.")
  730. return
  731. if increment_total_backfilled_portals:
  732. self.total_backfilled_portals = (self.total_backfilled_portals or 0) + 1
  733. await self.update()
  734. if local_limit is None or found_thread_count < local_limit:
  735. if local_limit is None:
  736. self.log.info(
  737. "Reached end of thread list with no limit, marking thread sync as completed"
  738. )
  739. else:
  740. self.log.info(
  741. f"Reached end of thread list (got {found_thread_count} with "
  742. f"limit {local_limit}), marking thread sync as completed"
  743. )
  744. self.thread_sync_completed = True
  745. await self.update()
  746. async def run_with_sync_lock(self, func: Callable[[], Awaitable]):
  747. with self._sync_lock:
  748. retry_count = 0
  749. while retry_count < 5:
  750. try:
  751. retry_count += 1
  752. await func()
  753. # The sync was successful. Exit the loop.
  754. return
  755. except IGNotLoggedInError as e:
  756. await self.logout(error=e)
  757. return
  758. except Exception:
  759. self.log.exception(
  760. "Failed to sync threads. Waiting 30 seconds before retrying sync."
  761. )
  762. await asyncio.sleep(30)
  763. # If we get here, it means that the sync has failed five times. If this happens, most
  764. # likely something very bad has happened.
  765. self.log.error("Failed to sync threads five times. Will not retry.")
  766. def start_listen(self, is_after_sync: bool = False) -> None:
  767. self.shutdown = False
  768. task = self._listen(
  769. seq_id=self.seq_id, snapshot_at_ms=self.snapshot_at_ms, is_after_sync=is_after_sync
  770. )
  771. self._listen_task = self.loop.create_task(task)
  772. async def delayed_start_listen(self, sleep: int) -> None:
  773. await asyncio.sleep(sleep)
  774. if self.is_connected:
  775. self.log.debug(
  776. "Already reconnected before delay after MQTT reconnection error finished"
  777. )
  778. else:
  779. self.log.debug("Reconnecting after MQTT connection error")
  780. self.start_listen()
  781. async def fetch_user_and_reconnect(self, sleep_first: int | None = None) -> None:
  782. if sleep_first:
  783. await asyncio.sleep(sleep_first)
  784. if self.is_connected:
  785. self.log.debug("Canceling user fetch, already reconnected")
  786. return
  787. self.log.debug("Refetching current user after disconnection")
  788. errors = 0
  789. while True:
  790. try:
  791. resp = await self.client.current_user()
  792. except (
  793. ProxyError,
  794. ProxyTimeoutError,
  795. ProxyConnectionError,
  796. ClientConnectionError,
  797. ConnectionError,
  798. asyncio.TimeoutError,
  799. ) as e:
  800. errors += 1
  801. wait = min(errors * 10, 60)
  802. self.log.warning(
  803. f"{e.__class__.__name__} while trying to check user for reconnection, "
  804. f"retrying in {wait} seconds: {e}"
  805. )
  806. await asyncio.sleep(wait)
  807. await self._maybe_update_proxy("fetch_user_and_reconnect error")
  808. except IGNotLoggedInError as e:
  809. self.log.warning(f"Failed to reconnect to Instagram: {e}, logging out")
  810. await self.logout(error=e)
  811. return
  812. except (IGChallengeError, IGConsentRequiredError) as e:
  813. await self._handle_checkpoint(e, on="reconnect")
  814. return
  815. except Exception as e:
  816. self.log.exception("Error while reconnecting to Instagram")
  817. if isinstance(e, IGCheckpointError):
  818. self.log.debug("Checkpoint error content: %s", e.body)
  819. await self.push_bridge_state(
  820. BridgeStateEvent.UNKNOWN_ERROR, info={"python_error": str(e)}
  821. )
  822. return
  823. else:
  824. self.log.debug(f"Confirmed current user {resp.user.pk}")
  825. self.start_listen()
  826. return
  827. async def _listen(self, seq_id: int, snapshot_at_ms: int, is_after_sync: bool) -> None:
  828. try:
  829. await self.mqtt.listen(
  830. graphql_subs={
  831. GraphQLSubscription.app_presence(),
  832. GraphQLSubscription.direct_typing(self.state.user_id),
  833. GraphQLSubscription.direct_status(),
  834. },
  835. skywalker_subs={
  836. SkywalkerSubscription.direct_sub(self.state.user_id),
  837. SkywalkerSubscription.live_sub(self.state.user_id),
  838. },
  839. seq_id=seq_id,
  840. snapshot_at_ms=snapshot_at_ms,
  841. )
  842. except IrisSubscribeError as e:
  843. if is_after_sync:
  844. self.log.exception("Got IrisSubscribeError right after refresh")
  845. await self.send_bridge_notice(
  846. f"Reconnecting failed again after refresh: {e}",
  847. important=True,
  848. state_event=BridgeStateEvent.UNKNOWN_ERROR,
  849. error_code="ig-refresh-connection-error",
  850. error_message=str(e),
  851. info={"python_error": str(e)},
  852. )
  853. else:
  854. self.log.warning(f"Got IrisSubscribeError {e}, refreshing...")
  855. asyncio.create_task(self.refresh())
  856. except MQTTReconnectionError as e:
  857. self.log.warning(
  858. f"Unexpected connection error: {e}, reconnecting in 1 minute", exc_info=True
  859. )
  860. await self.send_bridge_notice(
  861. f"Error in listener: {e}",
  862. important=True,
  863. state_event=BridgeStateEvent.TRANSIENT_DISCONNECT,
  864. error_code="ig-connection-error-socket",
  865. )
  866. self.mqtt.disconnect()
  867. asyncio.create_task(self.delayed_start_listen(sleep=60))
  868. except (MQTTNotConnected, MQTTNotLoggedIn, MQTTConnectionUnauthorized) as e:
  869. self.log.warning(f"Unexpected connection error: {e}, checking auth and reconnecting")
  870. await self.send_bridge_notice(
  871. f"Error in listener: {e}",
  872. important=True,
  873. state_event=BridgeStateEvent.UNKNOWN_ERROR,
  874. error_code="ig-connection-error-maybe-auth",
  875. )
  876. self.mqtt.disconnect()
  877. asyncio.create_task(self.fetch_user_and_reconnect())
  878. except Exception as e:
  879. self.log.exception("Fatal error in listener, reconnecting in 5 minutes")
  880. await self.send_bridge_notice(
  881. "Fatal error in listener (see logs for more info)",
  882. state_event=BridgeStateEvent.UNKNOWN_ERROR,
  883. important=True,
  884. error_code="ig-unknown-connection-error",
  885. info={"python_error": str(e)},
  886. )
  887. self.mqtt.disconnect()
  888. asyncio.create_task(self.fetch_user_and_reconnect(sleep_first=300))
  889. else:
  890. if not self.shutdown:
  891. await self.send_bridge_notice(
  892. "Instagram connection closed without error",
  893. state_event=BridgeStateEvent.UNKNOWN_ERROR,
  894. error_code="ig-disconnected",
  895. )
  896. finally:
  897. self._listen_task = None
  898. self._is_connected = False
  899. self._track_metric(METRIC_CONNECTED, False)
  900. async def stop_listen(self) -> None:
  901. if self.mqtt:
  902. self.shutdown = True
  903. self.mqtt.disconnect()
  904. if self._listen_task:
  905. await self._listen_task
  906. self.shutdown = False
  907. self._track_metric(METRIC_CONNECTED, False)
  908. self._is_connected = False
  909. await self.update()
  910. def stop_backfill_tasks(self) -> None:
  911. if self._backfill_loop_task:
  912. self._backfill_loop_task.cancel()
  913. self._backfill_loop_task = None
  914. if self._thread_sync_task:
  915. self._thread_sync_task.cancel()
  916. self._thread_sync_task = None
  917. async def logout(self, error: IGNotLoggedInError | None = None) -> None:
  918. await self.stop_listen()
  919. self.stop_backfill_tasks()
  920. if self.client and error is None:
  921. try:
  922. await self.client.logout(one_tap_app_login=False)
  923. except Exception:
  924. self.log.debug("Exception logging out", exc_info=True)
  925. if self.mqtt:
  926. self.mqtt.disconnect()
  927. self._track_metric(METRIC_CONNECTED, False)
  928. self._track_metric(METRIC_LOGGED_IN, False)
  929. if error is None:
  930. await self.push_bridge_state(BridgeStateEvent.LOGGED_OUT)
  931. puppet = await pu.Puppet.get_by_pk(self.igpk, create=False)
  932. if puppet and puppet.is_real_user:
  933. await puppet.switch_mxid(None, None)
  934. try:
  935. del self.by_igpk[self.igpk]
  936. except KeyError:
  937. pass
  938. self.igpk = None
  939. else:
  940. self.log.debug("Auth error body: %s", error.body.serialize())
  941. await self.send_bridge_notice(
  942. f"You have been logged out of Instagram: {error.proper_message}",
  943. important=True,
  944. state_event=BridgeStateEvent.BAD_CREDENTIALS,
  945. error_code="ig-auth-error",
  946. error_message=error.proper_message,
  947. )
  948. self.client = None
  949. self.mqtt = None
  950. self.state = None
  951. self.seq_id = None
  952. self.snapshot_at_ms = None
  953. self.thread_sync_completed = False
  954. self._is_logged_in = False
  955. await self.update()
  956. # endregion
  957. # region Event handlers
  958. async def _save_seq_id_after_sleep(self) -> None:
  959. await asyncio.sleep(120)
  960. self._seq_id_save_task = None
  961. self.log.trace("Saving sequence ID %d/%d", self.seq_id, self.snapshot_at_ms)
  962. try:
  963. await self.save_seq_id()
  964. except Exception:
  965. self.log.exception("Error saving sequence ID")
  966. async def update_seq_id(self, evt: NewSequenceID) -> None:
  967. self.seq_id = evt.seq_id
  968. self.snapshot_at_ms = evt.snapshot_at_ms
  969. if not self._seq_id_save_task or self._seq_id_save_task.done():
  970. self.log.trace("Starting seq id save task (%d/%d)", evt.seq_id, evt.snapshot_at_ms)
  971. self._seq_id_save_task = asyncio.create_task(self._save_seq_id_after_sleep())
  972. else:
  973. self.log.trace("Not starting seq id save task (%d/%d)", evt.seq_id, evt.snapshot_at_ms)
  974. @async_time(METRIC_MESSAGE)
  975. async def handle_message(self, evt: MessageSyncEvent) -> None:
  976. portal = await po.Portal.get_by_thread_id(evt.message.thread_id, receiver=self.igpk)
  977. if not portal or not portal.mxid:
  978. self.log.debug("Got message in thread with no portal, getting info...")
  979. resp = await self.client.get_thread(evt.message.thread_id)
  980. portal = await po.Portal.get_by_thread(resp.thread, self.igpk)
  981. self.log.debug("Got info for unknown portal, creating room")
  982. await portal.create_matrix_room(self, resp.thread)
  983. if not portal.mxid:
  984. self.log.warning(
  985. "Room creation appears to have failed, "
  986. f"dropping message in {evt.message.thread_id}"
  987. )
  988. return
  989. self.log.trace(f"Received message sync event {evt.message}")
  990. if evt.message.new_reaction:
  991. await portal.handle_instagram_reaction(
  992. evt.message, remove=evt.message.op == Operation.REMOVE
  993. )
  994. return
  995. sender = await pu.Puppet.get_by_pk(evt.message.user_id) if evt.message.user_id else None
  996. if evt.message.op == Operation.ADD:
  997. if not sender:
  998. # I don't think we care about adds with no sender
  999. return
  1000. await portal.handle_instagram_item(self, sender, evt.message)
  1001. elif evt.message.op == Operation.REMOVE:
  1002. # Removes don't have a sender, only the message sender can unsend messages anyway
  1003. await portal.handle_instagram_remove(evt.message.item_id)
  1004. elif evt.message.op == Operation.REPLACE:
  1005. await portal.handle_instagram_update(evt.message)
  1006. @async_time(METRIC_THREAD_SYNC)
  1007. async def handle_thread_sync(self, evt: ThreadSyncEvent) -> None:
  1008. self.log.trace("Thread sync event content: %s", evt)
  1009. portal = await po.Portal.get_by_thread(evt, receiver=self.igpk)
  1010. if portal.mxid:
  1011. self.log.debug("Got thread sync event for %s with existing portal", portal.thread_id)
  1012. await portal.update_matrix_room(self, evt)
  1013. elif evt.is_group:
  1014. self.log.debug(
  1015. "Got thread sync event for group %s without existing portal, creating room",
  1016. portal.thread_id,
  1017. )
  1018. await portal.create_matrix_room(self, evt)
  1019. else:
  1020. self.log.debug(
  1021. "Got thread sync event for DM %s without existing portal, ignoring",
  1022. portal.thread_id,
  1023. )
  1024. async def handle_thread_remove(self, evt: ThreadRemoveEvent) -> None:
  1025. self.log.debug("Got thread remove event: %s", evt.serialize())
  1026. @async_time(METRIC_RTD)
  1027. async def handle_rtd(self, evt: RealtimeDirectEvent) -> None:
  1028. if not isinstance(evt.value, ActivityIndicatorData):
  1029. return
  1030. now = int(time.time() * 1000)
  1031. date = evt.value.timestamp_ms
  1032. expiry = date + evt.value.ttl
  1033. if expiry < now:
  1034. return
  1035. if evt.activity_indicator_id in self._activity_indicator_ids:
  1036. return
  1037. # TODO clear expired items from this dict
  1038. self._activity_indicator_ids[evt.activity_indicator_id] = expiry
  1039. puppet = await pu.Puppet.get_by_pk(int(evt.value.sender_id))
  1040. portal = await po.Portal.get_by_thread_id(evt.thread_id, receiver=self.igpk)
  1041. if not puppet or not portal or not portal.mxid:
  1042. return
  1043. is_typing = evt.value.activity_status != TypingStatus.OFF
  1044. if puppet.pk == self.igpk:
  1045. self.remote_typing_status = TypingStatus.TEXT if is_typing else TypingStatus.OFF
  1046. await puppet.intent_for(portal).set_typing(portal.mxid, timeout=evt.value.ttl)
  1047. # endregion
  1048. # region Database getters
  1049. def _add_to_cache(self) -> None:
  1050. self.by_mxid[self.mxid] = self
  1051. if self.igpk:
  1052. self.by_igpk[self.igpk] = self
  1053. @classmethod
  1054. @async_getter_lock
  1055. async def get_by_mxid(cls, mxid: UserID, *, create: bool = True) -> User | None:
  1056. # Never allow ghosts to be users
  1057. if pu.Puppet.get_id_from_mxid(mxid):
  1058. return None
  1059. try:
  1060. return cls.by_mxid[mxid]
  1061. except KeyError:
  1062. pass
  1063. user = cast(cls, await super().get_by_mxid(mxid))
  1064. if user is not None:
  1065. user._add_to_cache()
  1066. return user
  1067. if create:
  1068. user = cls(mxid)
  1069. await user.insert()
  1070. user._add_to_cache()
  1071. return user
  1072. return None
  1073. @classmethod
  1074. @async_getter_lock
  1075. async def get_by_igpk(cls, igpk: int) -> User | None:
  1076. try:
  1077. return cls.by_igpk[igpk]
  1078. except KeyError:
  1079. pass
  1080. user = cast(cls, await super().get_by_igpk(igpk))
  1081. if user is not None:
  1082. user._add_to_cache()
  1083. return user
  1084. return None
  1085. @classmethod
  1086. async def all_logged_in(cls) -> AsyncGenerator[User, None]:
  1087. users = await super().all_logged_in()
  1088. user: cls
  1089. for index, user in enumerate(users):
  1090. try:
  1091. yield cls.by_mxid[user.mxid]
  1092. except KeyError:
  1093. user._add_to_cache()
  1094. yield user
  1095. # endregion