mautrix-instagram/mauigpapi/mqtt/conn.py

# mautrix-instagram - A Matrix-Instagram puppeting bridge.
# Copyright (C) 2022 Tulir Asokan
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.
from __future__ import annotations

from typing import Any, Awaitable, Callable, Iterable, Type, TypeVar
from collections import defaultdict
from socket import error as SocketError, socket
import asyncio
import json
import logging
import re
import time
import zlib

from yarl import URL
import paho.mqtt.client as pmc

from mautrix.util import background_task
from mautrix.util.logging import TraceLogger
from mautrix.util.proxy import ProxyHandler, proxy_with_retry

from ..errors import (
    IrisSubscribeError,
    MQTTConnectionUnauthorized,
    MQTTNotConnected,
    MQTTNotLoggedIn,
    MQTTReconnectionError,
)
from ..state import AndroidState
from ..types import (
    AppPresenceEventPayload,
    ClientConfigUpdatePayload,
    CommandResponse,
    IrisPayload,
    IrisPayloadData,
    LiveVideoCommentPayload,
    MessageSyncEvent,
    MessageSyncMessage,
    Operation,
    PubsubEvent,
    PubsubPayload,
    ReactionStatus,
    ReactionType,
    RealtimeDirectEvent,
    RealtimeZeroProvisionPayload,
    ThreadAction,
    ThreadItemType,
    ThreadRemoveEvent,
    ThreadSyncEvent,
    TypingStatus,
)
from .events import Connect, Disconnect, NewSequenceID, ProxyUpdate
from .otclient import MQTToTClient
from .subscription import GraphQLQueryID, RealtimeTopic, everclear_subscriptions
from .thrift import ForegroundStateConfig, IncomingMessage, RealtimeClientInfo, RealtimeConfig

try:
    import socks
except ImportError:
    socks = None

T = TypeVar("T")

ACTIVITY_INDICATOR_REGEX = re.compile(
    r"/direct_v2/threads/([\w_]+)/activity_indicator_id/([\w_]+)"
)

INBOX_THREAD_REGEX = re.compile(r"/direct_v2/inbox/threads/([\w_]+)")

REQUEST_TIMEOUT = 60 * 3
REQUEST_KEEPALIVE = 5


class AndroidMQTT:
    _loop: asyncio.AbstractEventLoop
    _client: MQTToTClient
    log: TraceLogger
    state: AndroidState
    _graphql_subs: set[str]
    _skywalker_subs: set[str]
    _iris_seq_id: int | None
    _iris_snapshot_at_ms: int | None
    _publish_waiters: dict[int, asyncio.Future]
    _response_waiters: dict[RealtimeTopic, asyncio.Future]
    _response_waiter_locks: dict[RealtimeTopic, asyncio.Lock]
    _message_response_waiter_lock: asyncio.Lock
    _message_response_waiter_id: str | None
    _message_response_waiter: asyncio.Future | None
    _disconnect_error: Exception | None
    _event_handlers: dict[Type[T], list[Callable[[T], Awaitable[None]]]]
    _outgoing_events: asyncio.Queue
    _event_dispatcher_task: asyncio.Task | None

    # region Initialization

    def __init__(
        self,
        state: AndroidState,
        log: TraceLogger | None = None,
        proxy_handler: ProxyHandler | None = None,
        mqtt_keepalive: int = 60,
    ) -> None:
        self._graphql_subs = set()
        self._skywalker_subs = set()
        self._iris_seq_id = None
        self._iris_snapshot_at_ms = None
        self._publish_waiters = {}
        self._response_waiters = {}
        self._message_response_waiter_lock = asyncio.Lock()
        self._message_response_waiter_id = None
        self._message_response_waiter = None
        self._disconnect_error = None
        self._response_waiter_locks = defaultdict(lambda: asyncio.Lock())
        self._event_handlers = defaultdict(lambda: [])
        self._event_dispatcher_task = None
        self._outgoing_events = asyncio.Queue()
        self.log = log or logging.getLogger("mauigpapi.mqtt")
        self._loop = asyncio.get_running_loop()
        self.state = state
        self._client = MQTToTClient(
            client_id=self._form_client_id(),
            clean_session=True,
            protocol=pmc.MQTTv31,
            transport="tcp",
        )
        self.proxy_handler = proxy_handler
        self.setup_proxy()
        self._client.enable_logger()
        self._client.tls_set()
        # mqtt.max_inflight_messages_set(20)  # The rest will get queued
        # mqtt.max_queued_messages_set(0)  # Unlimited messages can be queued
        # mqtt.message_retry_set(20)  # Retry sending for at least 20 seconds
        # mqtt.reconnect_delay_set(min_delay=1, max_delay=120)
        self._default_keepalive = mqtt_keepalive
        self._client.connect_async("edge-mqtt.facebook.com", 443, keepalive=mqtt_keepalive)
        self._client.on_message = self._on_message_handler
        self._client.on_publish = self._on_publish_handler
        self._client.on_connect = self._on_connect_handler
        self._client.on_disconnect = self._on_disconnect_handler
        self._client.on_socket_open = self._on_socket_open
        self._client.on_socket_close = self._on_socket_close
        self._client.on_socket_register_write = self._on_socket_register_write
        self._client.on_socket_unregister_write = self._on_socket_unregister_write

    def setup_proxy(self):
        http_proxy = self.proxy_handler.get_proxy_url() if self.proxy_handler else None
        if http_proxy:
            if not socks:
                self.log.warning("http_proxy is set, but pysocks is not installed")
            else:
                proxy_url = URL(http_proxy)
                proxy_type = {
                    "http": socks.HTTP,
                    "https": socks.HTTP,
                    "socks": socks.SOCKS5,
                    "socks5": socks.SOCKS5,
                    "socks4": socks.SOCKS4,
                }[proxy_url.scheme]
                self._client.proxy_set(
                    proxy_type=proxy_type,
                    proxy_addr=proxy_url.host,
                    proxy_port=proxy_url.port,
                    proxy_username=proxy_url.user,
                    proxy_password=proxy_url.password,
                )

    def _clear_publish_waiters(self) -> None:
        for waiter in self._publish_waiters.values():
            if not waiter.done():
                waiter.set_exception(MQTTNotConnected("MQTT disconnected before PUBACK received"))
        self._publish_waiters = {}

    def _form_client_id(self) -> bytes:
        subscribe_topics = [
            RealtimeTopic.PUBSUB,  # 88
            RealtimeTopic.SUB_IRIS_RESPONSE,  # 135
            RealtimeTopic.REALTIME_SUB,  # 149
            RealtimeTopic.REGION_HINT,  # 150
            RealtimeTopic.T_RTC_LOG,  # 274
            RealtimeTopic.SEND_MESSAGE_RESPONSE,  # 133
            RealtimeTopic.LARGE_SCALE_FIRE_AND_FORGET_SYNC,  # 279
            RealtimeTopic.MESSAGE_SYNC,  # 146
            RealtimeTopic.LIGHTSPEED_RESPONSE,  # 179
            RealtimeTopic.UNKNOWN_PP,  # 34
        ]
        subscribe_topic_ids = [int(topic.encoded) for topic in subscribe_topics]
        password = f"authorization={self.state.session.authorization}"
        cfg = RealtimeConfig(
            client_identifier=self.state.device.phone_id[:20],
            client_info=RealtimeClientInfo(
                user_id=int(self.state.user_id),
                user_agent=self.state.user_agent,
                client_capabilities=0b10110111,
                endpoint_capabilities=0,
                publish_format=1,
                no_automatic_foreground=True,
                make_user_available_in_foreground=False,
                device_id=self.state.device.phone_id,
                is_initially_foreground=False,
                network_type=1,
                network_subtype=-1,
                client_mqtt_session_id=int(time.time() * 1000) & 0xFFFFFFFF,
                subscribe_topics=subscribe_topic_ids,
                client_type="cookie_auth",
                app_id=567067343352427,
                region_preference=self.state.session.region_hint or "LLA",
                device_secret="",
                client_stack=3,
            ),
            password=password,
            app_specific_info={
                "capabilities": self.state.application.CAPABILITIES,
                "app_version": self.state.application.APP_VERSION,
                "everclear_subscriptions": json.dumps(everclear_subscriptions),
                "User-Agent": self.state.user_agent,
                "Accept-Language": self.state.device.language.replace("_", "-"),
                "platform": "android",
                "ig_mqtt_route": "django",
                "pubsub_msg_type_blacklist": "direct, typing_type",
                # "auth_cache_enabled": "1",
            },
        )
        return zlib.compress(cfg.to_thrift(), level=9)

    # endregion

    def _on_socket_open(self, client: MQTToTClient, _: Any, sock: socket) -> None:
        self._loop.add_reader(sock, client.loop_read)

    def _on_socket_close(self, client: MQTToTClient, _: Any, sock: socket) -> None:
        self._loop.remove_reader(sock)

    def _on_socket_register_write(self, client: MQTToTClient, _: Any, sock: socket) -> None:
        self._loop.add_writer(sock, client.loop_write)

    def _on_socket_unregister_write(self, client: MQTToTClient, _: Any, sock: socket) -> None:
        self._loop.remove_writer(sock)

    def _on_connect_handler(
        self, client: MQTToTClient, _: Any, flags: dict[str, Any], rc: int
    ) -> None:
        if rc != 0:
            err = pmc.connack_string(rc)
            self.log.error("MQTT Connection Error: %s (%d)", err, rc)
            if rc == pmc.CONNACK_REFUSED_NOT_AUTHORIZED:
                self._disconnect_error = MQTTConnectionUnauthorized()
                self.disconnect()
            return

        self._loop.create_task(self._post_connect())

    def _on_disconnect_handler(self, client: MQTToTClient, _: Any, rc: int) -> None:
        err_str = "Generic error." if rc == pmc.MQTT_ERR_NOMEM else pmc.error_string(rc)
        self.log.debug("MQTT disconnection code %d: %s", rc, err_str)
        self._clear_publish_waiters()

    async def _post_connect(self) -> None:
        await self._dispatch(Connect())
        self.log.debug("Re-subscribing to things after connect")
        if self._graphql_subs:
            res = await self.graphql_subscribe(self._graphql_subs)
            self.log.trace("GraphQL subscribe response: %s", res)
        if self._skywalker_subs:
            res = await self.skywalker_subscribe(self._skywalker_subs)
            self.log.trace("Skywalker subscribe response: %s", res)
        if self._iris_seq_id:
            retry = 0
            while True:
                try:
                    await self.iris_subscribe(self._iris_seq_id, self._iris_snapshot_at_ms)
                    break
                except (asyncio.TimeoutError, IrisSubscribeError) as e:
                    self.log.exception("Error requesting iris subscribe")
                    retry += 1
                    if retry >= 5 or isinstance(e, IrisSubscribeError):
                        self._disconnect_error = e
                        self.disconnect()
                        break
                    await asyncio.sleep(5)
                    self.log.debug("Retrying iris subscribe")

    def _on_publish_handler(self, client: MQTToTClient, _: Any, mid: int) -> None:
        try:
            waiter = self._publish_waiters.pop(mid)
        except KeyError:
            self.log.trace(f"Got publish confirmation for {mid}, but no waiters")
            return
        self.log.trace(f"Got publish confirmation for {mid}")
        if not waiter.done():
            waiter.set_result(None)
        self.maybe_reset_keepalive()

    # region Incoming event parsing

    def _parse_direct_thread_path(self, path: str) -> dict:
        try:
            blank, direct_v2, threads, thread_id, *rest = path.split("/")
        except (ValueError, IndexError) as e:
            self.log.debug(f"Got {e!r} while parsing path {path}")
            raise
        if (blank, direct_v2, threads) != ("", "direct_v2", "threads"):
            self.log.debug(f"Got unexpected first parts in direct thread path {path}")
            raise ValueError("unexpected first three parts in _parse_direct_thread_path")
        additional = {"thread_id": thread_id}
        if rest:
            subitem_key = rest[0]
            if subitem_key == "approval_required_for_new_members":
                additional["approval_required_for_new_members"] = True
            elif subitem_key == "thread_image":
                additional["is_thread_image"] = True
            elif subitem_key == "participants" and len(rest) > 2 and rest[2] == "has_seen":
                additional["has_seen"] = int(rest[1])
            elif subitem_key == "items":
                additional["item_id"] = rest[1]
                if len(rest) > 4 and rest[2] == "reactions":
                    additional["reaction_type"] = ReactionType(rest[3])
                    additional["reaction_user_id"] = int(rest[4])
            elif subitem_key in "admin_user_ids":
                additional["admin_user_id"] = int(rest[1])
            elif subitem_key == "activity_indicator_id":
                additional["activity_indicator_id"] = rest[1]
        self.log.trace("Parsed path %s -> %s", path, additional)
        return additional

    def _on_messager_sync_item(self, part: IrisPayloadData, parsed_item: IrisPayload) -> bool:
        if part.path.startswith("/direct_v2/threads/"):
            raw_message = {
                "path": part.path,
                "op": part.op,
                **self._parse_direct_thread_path(part.path),
            }
            try:
                json_value = json.loads(part.value)
                if "reaction_type" in raw_message:
                    self.log.trace("Treating %s as new reaction data", json_value)
                    raw_message["new_reaction"] = json_value
                    json_value["sender_id"] = raw_message.pop("reaction_user_id")
                    json_value["type"] = raw_message.pop("reaction_type")
                    json_value["client_context"] = parsed_item.mutation_token
                    if part.op == Operation.REMOVE:
                        json_value["emoji"] = None
                        json_value["timestamp"] = None
                else:
                    raw_message = {
                        **raw_message,
                        **json_value,
                    }
            except (json.JSONDecodeError, TypeError):
                raw_message["value"] = part.value
            message = MessageSyncMessage.deserialize(raw_message)
            evt = MessageSyncEvent(iris=parsed_item, message=message)
        elif part.path.startswith("/direct_v2/inbox/threads/"):
            if part.op == Operation.REMOVE:
                blank, direct_v2, inbox, threads, thread_id, *_ = part.path.split("/")
                evt = ThreadRemoveEvent.deserialize(
                    {
                        "thread_id": thread_id,
                        "path": part.path,
                        "op": part.op,
                        **json.loads(part.value),
                    }
                )
            else:
                evt = ThreadSyncEvent.deserialize(
                    {
                        "path": part.path,
                        "op": part.op,
                        **json.loads(part.value),
                    }
                )
        else:
            self.log.warning(f"Unsupported path {part.path}")
            return False
        self._outgoing_events.put_nowait(evt)
        return True

    def _on_message_sync(self, payload: bytes) -> None:
        parsed = json.loads(payload.decode("utf-8"))
        self.log.trace("Got message sync event: %s", parsed)
        has_items = False
        for sync_item in parsed:
            parsed_item = IrisPayload.deserialize(sync_item)
            if self._iris_seq_id < parsed_item.seq_id:
                self.log.trace(f"Got new seq_id: {parsed_item.seq_id}")
                self._iris_seq_id = parsed_item.seq_id
                self._iris_snapshot_at_ms = int(time.time() * 1000)
                background_task.create(
                    self._dispatch(NewSequenceID(self._iris_seq_id, self._iris_snapshot_at_ms))
                )
            for part in parsed_item.data:
                has_items = self._on_messager_sync_item(part, parsed_item) or has_items
        if has_items and not self._event_dispatcher_task:
            self._event_dispatcher_task = asyncio.create_task(self._dispatcher_loop())

    def _on_pubsub(self, payload: bytes) -> None:
        parsed_thrift = IncomingMessage.from_thrift(payload)
        self.log.trace(f"Got pubsub event {parsed_thrift.topic} / {parsed_thrift.payload}")
        message = PubsubPayload.parse_json(parsed_thrift.payload)
        for data in message.data:
            match = ACTIVITY_INDICATOR_REGEX.match(data.path)
            if match:
                evt = PubsubEvent(
                    data=data,
                    base=message,
                    thread_id=match.group(1),
                    activity_indicator_id=match.group(2),
                )
                self._loop.create_task(self._dispatch(evt))
            elif not data.double_publish:
                self.log.debug("Pubsub: no activity indicator on data: %s", data)
            else:
                self.log.debug("Pubsub: double publish: %s", data.path)

    def _parse_realtime_sub_item(self, topic: str | GraphQLQueryID, raw: dict) -> Iterable[Any]:
        if topic == GraphQLQueryID.APP_PRESENCE:
            yield AppPresenceEventPayload.deserialize(raw).presence_event
        elif topic == GraphQLQueryID.ZERO_PROVISION:
            yield RealtimeZeroProvisionPayload.deserialize(raw).zero_product_provisioning_event
        elif topic == GraphQLQueryID.CLIENT_CONFIG_UPDATE:
            yield ClientConfigUpdatePayload.deserialize(raw).client_config_update_event
        elif topic == GraphQLQueryID.LIVE_REALTIME_COMMENTS:
            yield LiveVideoCommentPayload.deserialize(raw).live_video_comment_event
        elif topic == "direct":
            event = raw["event"]
            for item in raw["data"]:
                yield RealtimeDirectEvent.deserialize(
                    {
                        "event": event,
                        **self._parse_direct_thread_path(item["path"]),
                        **item,
                    }
                )

    def _on_realtime_sub(self, payload: bytes) -> None:
        parsed_thrift = IncomingMessage.from_thrift(payload)
        try:
            topic = GraphQLQueryID(parsed_thrift.topic)
        except ValueError:
            topic = parsed_thrift.topic
        self.log.trace(f"Got realtime sub event {topic} / {parsed_thrift.payload}")
        allowed = (
            "direct",
            GraphQLQueryID.APP_PRESENCE,
            GraphQLQueryID.ZERO_PROVISION,
            GraphQLQueryID.CLIENT_CONFIG_UPDATE,
            GraphQLQueryID.LIVE_REALTIME_COMMENTS,
        )
        if topic not in allowed:
            return
        parsed_json = json.loads(parsed_thrift.payload)
        for evt in self._parse_realtime_sub_item(topic, parsed_json):
            self._loop.create_task(self._dispatch(evt))

    def _handle_send_response(self, message: pmc.MQTTMessage) -> None:
        data = json.loads(message.payload.decode("utf-8"))
        try:
            ccid = data["payload"]["client_context"]
        except KeyError:
            self.log.warning(
                "Didn't find client_context in send message response: %s", message.payload
            )
            ccid = self._message_response_waiter_id
        else:
            if ccid != self._message_response_waiter_id:
                self.log.error(
                    "Mismatching client_context in send message response (%s != %s)",
                    ccid,
                    self._message_response_waiter_id,
                )
                return
        if self._message_response_waiter and not self._message_response_waiter.done():
            self.log.debug("Got response to %s: %s", ccid, message.payload)
            self._message_response_waiter.set_result(message)
            self._message_response_waiter = None
            self._message_response_waiter_id = None
        else:
            self.log.warning("Didn't find task waiting for response %s", message.payload)

    def _on_message_handler(self, client: MQTToTClient, _: Any, message: pmc.MQTTMessage) -> None:
        try:
            topic = RealtimeTopic.decode(message.topic)
            # Instagram Android MQTT messages are always compressed
            message.payload = zlib.decompress(message.payload)
            if topic == RealtimeTopic.MESSAGE_SYNC:
                self._on_message_sync(message.payload)
            elif topic == RealtimeTopic.PUBSUB:
                self._on_pubsub(message.payload)
            elif topic == RealtimeTopic.REALTIME_SUB:
                self._on_realtime_sub(message.payload)
            elif topic == RealtimeTopic.SEND_MESSAGE_RESPONSE:
                self._handle_send_response(message)
            elif topic == RealtimeTopic.UNKNOWN_PP:
                self.log.warning("Reconnecting after receiving /pp message")
                background_task.create(self._reconnect())
            else:
                try:
                    waiter = self._response_waiters.pop(topic)
                except KeyError:
                    self.log.debug(
                        "No handler for MQTT message in %s: %s", topic.value, message.payload
                    )
                else:
                    if not waiter.done():
                        waiter.set_result(message)
                        self.log.trace("Got response %s: %s", topic.value, message.payload)
                    else:
                        self.log.debug(
                            "Got response in %s, but waiter was already cancelled: %s",
                            topic,
                            message.payload,
                        )
        except Exception:
            self.log.exception("Error in incoming MQTT message handler")
            self.log.trace("Errored MQTT payload: %s", message.payload)
        finally:
            self.maybe_reset_keepalive()

    # endregion

    async def _reconnect(self) -> None:
        try:
            self._client.reconnect()
            return
        except (SocketError, OSError, pmc.WebsocketConnectionError) as e:
            self.log.exception("Error reconnecting to MQTT")
            raise MQTTReconnectionError("MQTT reconnection failed") from e

    def add_event_handler(
        self, evt_type: Type[T], handler: Callable[[T], Awaitable[None]]
    ) -> None:
        self._event_handlers[evt_type].append(handler)

    async def _dispatch(self, evt: T) -> None:
        for handler in self._event_handlers[type(evt)]:
            try:
                await handler(evt)
            except Exception:
                self.log.exception(f"Error in {type(evt).__name__} handler")

    def disconnect(self) -> None:
        self._client.disconnect()

    async def _dispatcher_loop(self) -> None:
        loop_id = f"{hex(id(self))}#{time.monotonic()}"
        self.log.debug(f"Dispatcher loop {loop_id} starting")
        try:
            while True:
                evt = await self._outgoing_events.get()
                await asyncio.shield(self._dispatch(evt))
        except asyncio.CancelledError:
            tasks = self._outgoing_events
            self._outgoing_events = asyncio.Queue()
            if not tasks.empty():
                self.log.debug(
                    f"Dispatcher loop {loop_id} stopping after dispatching {tasks.qsize()} events"
                )
            while not tasks.empty():
                await self._dispatch(tasks.get_nowait())
            raise
        finally:
            self.log.debug(f"Dispatcher loop {loop_id} stopped")

    async def listen(
        self,
        graphql_subs: set[str] | None = None,
        skywalker_subs: set[str] | None = None,
        seq_id: int = None,
        snapshot_at_ms: int = None,
        retry_limit: int = 10,
    ) -> None:
        self._graphql_subs = graphql_subs or set()
        self._skywalker_subs = skywalker_subs or set()
        self._iris_seq_id = seq_id
        self._iris_snapshot_at_ms = snapshot_at_ms

        self.log.debug("Connecting to Instagram MQTT")

        async def connect_and_watch():
            await self._reconnect()

            while True:
                try:
                    await asyncio.sleep(1)
                except asyncio.CancelledError:
                    self.disconnect()
                    # this might not be necessary
                    self._client.loop_misc()
                    return
                rc = self._client.loop_misc()

                # If disconnect() has been called
                # Beware, internal API, may have to change this to something more stable!
                if self._client._state == pmc.mqtt_cs_disconnecting:
                    return  # Stop listening

                if rc != pmc.MQTT_ERR_SUCCESS:
                    # If known/expected error
                    if rc == pmc.MQTT_ERR_CONN_LOST:
                        await self._dispatch(Disconnect(reason="Connection lost, retrying"))
                        raise MQTTNotConnected("MQTT_ERR_CONN_LOST")
                    elif rc == pmc.MQTT_ERR_NOMEM:
                        # This error is wrongly classified
                        # See https://github.com/eclipse/paho.mqtt.python/issues/340
                        await self._dispatch(Disconnect(reason="Connection lost, retrying"))
                        raise MQTTNotConnected("MQTT_ERR_NOMEM")
                    elif rc == pmc.MQTT_ERR_CONN_REFUSED:
                        await self._dispatch(Disconnect(reason="Connection refused, retrying"))
                        raise MQTTNotLoggedIn("MQTT_ERR_CONN_REFUSED")
                    elif rc == pmc.MQTT_ERR_NO_CONN:
                        await self._dispatch(Disconnect(reason="Connection dropped, retrying"))
                        raise MQTTNotConnected("MQTT_ERR_NO_CONN")
                    else:
                        err = pmc.error_string(rc)
                        self.log.error("MQTT Error: %s", err)
                        await self._dispatch(Disconnect(reason=f"MQTT Error: {err}, retrying"))
                        raise MQTTNotConnected(err)

        await proxy_with_retry(
            "mqtt.listen",
            lambda: connect_and_watch(),
            logger=self.log,
            proxy_handler=self.proxy_handler,
            on_proxy_change=lambda: self._dispatch(ProxyUpdate()),
            max_retries=retry_limit,
            retryable_exceptions=(MQTTNotConnected, MQTTReconnectionError),
            # Wait 1s * errors, max 5s for fast reconnect or die
            max_wait_seconds=5,
            multiply_wait_seconds=1,
            # If connection stable for >1h, reset the error counter
            reset_after_seconds=3600,
        )

        if self._event_dispatcher_task:
            self._event_dispatcher_task.cancel()
            self._event_dispatcher_task = None
        if self._disconnect_error:
            self.log.info("disconnect_error is set, raising and clearing variable")
            err = self._disconnect_error
            self._disconnect_error = None
            raise err

    # region Basic outgoing MQTT

    @staticmethod
    def _publish_cancel_later(fut: asyncio.Future) -> None:
        if not fut.done():
            fut.set_exception(asyncio.TimeoutError("MQTT publish timed out"))

    @staticmethod
    def _request_cancel_later(fut: asyncio.Future) -> None:
        if not fut.done():
            fut.set_exception(asyncio.TimeoutError("MQTT request timed out"))

    # The following two functions mutate the client keepalive (cheeky) to temporarily increase
    # ping attempts during read/write to MQTT. If things are flowing this should change nothing,
    # as pings only send when idle. It should, however, allow the client to detect a bad MQTT
    # connection much quicker than the default keepalive.
    def set_request_keepalive(self):
        self._client._keepalive = REQUEST_KEEPALIVE

    def maybe_reset_keepalive(self):
        # Reset the keepalive back to the default value if we have no pending publish/receive
        if (
            not self._response_waiters
            and not self._publish_waiters
            and not self._message_response_waiter
        ):
            self._client._keepalive = self._default_keepalive

    def publish(self, topic: RealtimeTopic, payload: str | bytes | dict) -> asyncio.Future:
        if isinstance(payload, dict):
            payload = json.dumps(payload)
        if isinstance(payload, str):
            payload = payload.encode("utf-8")
        self.log.trace(f"Publishing message in {topic.value} ({topic.encoded}): {payload}")
        payload = zlib.compress(payload, level=9)
        self.set_request_keepalive()
        info = self._client.publish(topic.encoded, payload, qos=1)
        self.log.trace(f"Published message ID: {info.mid}")
        fut = self._loop.create_future()
        timeout_handle = self._loop.call_later(REQUEST_TIMEOUT, self._publish_cancel_later, fut)
        fut.add_done_callback(lambda _: timeout_handle.cancel())
        self._publish_waiters[info.mid] = fut
        return fut

    async def request(
        self,
        topic: RealtimeTopic,
        response: RealtimeTopic,
        payload: str | bytes | dict,
        timeout: int | None = None,
    ) -> pmc.MQTTMessage:
        async with self._response_waiter_locks[response]:
            fut = self._loop.create_future()
            self._response_waiters[response] = fut
            background_task.create(self.publish(topic, payload))
            self.log.trace(
                f"Request publish to {topic.value} queued, waiting for response {response.name}"
            )
            timeout_handle = self._loop.call_later(
                timeout or REQUEST_TIMEOUT, self._request_cancel_later, fut
            )
            fut.add_done_callback(lambda _: timeout_handle.cancel())
            return await fut

    async def iris_subscribe(self, seq_id: int, snapshot_at_ms: int) -> None:
        self.log.debug(f"Requesting iris subscribe {seq_id}/{snapshot_at_ms}")
        resp = await self.request(
            RealtimeTopic.SUB_IRIS,
            RealtimeTopic.SUB_IRIS_RESPONSE,
            {
                "seq_id": seq_id,
                "snapshot_at_ms": snapshot_at_ms,
                "snapshot_app_version": self.state.application.APP_VERSION,
                "timezone_offset": int(self.state.device.timezone_offset),
                "subscription_type": "message",
            },
            timeout=20,
        )
        self.log.debug("Iris subscribe response: %s", resp.payload.decode("utf-8"))
        resp_dict = json.loads(resp.payload.decode("utf-8"))
        if resp_dict["error_type"] and resp_dict["error_message"]:
            raise IrisSubscribeError(resp_dict["error_type"], resp_dict["error_message"])
        latest_seq_id = resp_dict.get("latest_seq_id")
        if latest_seq_id > self._iris_seq_id:
            self.log.info(f"Latest sequence ID is {latest_seq_id}, catching up from {seq_id}")
            self._iris_seq_id = latest_seq_id
            self._iris_snapshot_at_ms = resp_dict.get("subscribed_at_ms", int(time.time() * 1000))
            background_task.create(
                self._dispatch(NewSequenceID(self._iris_seq_id, self._iris_snapshot_at_ms))
            )

    def graphql_subscribe(self, subs: set[str]) -> asyncio.Future:
        self._graphql_subs |= subs
        return self.publish(RealtimeTopic.REALTIME_SUB, {"sub": list(subs)})

    def graphql_unsubscribe(self, subs: set[str]) -> asyncio.Future:
        self._graphql_subs -= subs
        return self.publish(RealtimeTopic.REALTIME_SUB, {"unsub": list(subs)})

    def skywalker_subscribe(self, subs: set[str]) -> asyncio.Future:
        self._skywalker_subs |= subs
        return self.publish(RealtimeTopic.PUBSUB, {"sub": list(subs)})

    def skywalker_unsubscribe(self, subs: set[str]) -> asyncio.Future:
        self._skywalker_subs -= subs
        return self.publish(RealtimeTopic.PUBSUB, {"unsub": list(subs)})

    # endregion
    # region Actually sending messages and stuff

    async def send_foreground_state(self, state: ForegroundStateConfig) -> None:
        self.log.debug("Updating foreground state: %s", state)
        await self.publish(
            RealtimeTopic.FOREGROUND_STATE, zlib.compress(state.to_thrift(), level=9)
        )
        if state.keep_alive_timeout:
            self._client._keepalive = state.keep_alive_timeout

    async def send_command(
        self,
        thread_id: str,
        action: ThreadAction,
        client_context: str | None = None,
        **kwargs: Any,
    ) -> CommandResponse | None:
        self.log.debug(f"Preparing to send {action} to {thread_id} with {client_context}")
        client_context = client_context or self.state.gen_client_context()
        req = {
            "thread_id": thread_id,
            "client_context": client_context,
            "offline_threading_id": client_context,
            "mutation_token": client_context,
            "action": action.value,
            "is_shh_mode": "0",
            "sampled": False,
            "session_id": f"UFS-{self.state.pigeon_session_id}-0",
            # "device_id": self.state.cookies["ig_did"],
            **kwargs,
            "btt_dual_send": False,
            "is_ae_dual_send": False,
        }
        lock_start = time.monotonic()
        async with self._message_response_waiter_lock:
            lock_wait_dur = time.monotonic() - lock_start
            if lock_wait_dur > 1:
                self.log.warning(f"Waited {lock_wait_dur:.3f} seconds to send {client_context}")
            fut = self._message_response_waiter = asyncio.Future()
            self._message_response_waiter_id = client_context
            background_task.create(self.publish(RealtimeTopic.SEND_MESSAGE, req))
            self.log.debug(
                f"Request publish to {RealtimeTopic.SEND_MESSAGE} queued, "
                f"waiting for response {RealtimeTopic.SEND_MESSAGE_RESPONSE}"
            )
            try:
                resp = await asyncio.wait_for(fut, timeout=REQUEST_TIMEOUT)
            except asyncio.TimeoutError:
                self.log.error(f"Request with ID {client_context} timed out!")
                raise
            return CommandResponse.parse_json(resp.payload.decode("utf-8"))

    def send_item(
        self,
        thread_id: str,
        item_type: ThreadItemType,
        shh_mode: bool = False,
        client_context: str | None = None,
        **kwargs: Any,
    ) -> Awaitable[CommandResponse]:
        return self.send_command(
            thread_id,
            item_type=item_type.value,
            is_shh_mode=str(int(shh_mode)),
            action=ThreadAction.SEND_ITEM,
            client_context=client_context,
            **kwargs,
        )

    def send_hashtag(
        self,
        thread_id: str,
        hashtag: str,
        text: str = "",
        shh_mode: bool = False,
        client_context: str | None = None,
    ) -> Awaitable[CommandResponse]:
        return self.send_item(
            thread_id,
            text=text,
            item_id=hashtag,
            shh_mode=shh_mode,
            item_type=ThreadItemType.HASHTAG,
            client_context=client_context,
        )

    def send_like(
        self, thread_id: str, shh_mode: bool = False, client_context: str | None = None
    ) -> Awaitable[CommandResponse]:
        return self.send_item(
            thread_id,
            shh_mode=shh_mode,
            item_type=ThreadItemType.LIKE,
            client_context=client_context,
        )

    def send_location(
        self,
        thread_id: str,
        venue_id: str,
        text: str = "",
        shh_mode: bool = False,
        client_context: str | None = None,
    ) -> Awaitable[CommandResponse]:
        return self.send_item(
            thread_id,
            text=text,
            item_id=venue_id,
            shh_mode=shh_mode,
            item_type=ThreadItemType.LOCATION,
            client_context=client_context,
        )

    def send_media(
        self,
        thread_id: str,
        media_id: str,
        text: str = "",
        shh_mode: bool = False,
        client_context: str | None = None,
    ) -> Awaitable[CommandResponse]:
        return self.send_item(
            thread_id,
            text=text,
            media_id=media_id,
            shh_mode=shh_mode,
            item_type=ThreadItemType.MEDIA_SHARE,
            client_context=client_context,
        )

    def send_profile(
        self,
        thread_id: str,
        user_id: str,
        text: str = "",
        shh_mode: bool = False,
        client_context: str | None = None,
    ) -> Awaitable[CommandResponse]:
        return self.send_item(
            thread_id,
            text=text,
            item_id=user_id,
            shh_mode=shh_mode,
            item_type=ThreadItemType.PROFILE,
            client_context=client_context,
        )

    def send_reaction(
        self,
        thread_id: str,
        emoji: str,
        item_id: str,
        reaction_status: ReactionStatus = ReactionStatus.CREATED,
        target_item_type: ThreadItemType = ThreadItemType.TEXT,
        shh_mode: bool = False,
        client_context: str | None = None,
        original_message_client_context: str | None = None,
    ) -> Awaitable[CommandResponse]:
        kwargs = (
            {
                "original_message_client_context": original_message_client_context,
            }
            if original_message_client_context
            else {}
        )
        return self.send_item(
            thread_id,
            reaction_status=reaction_status.value,
            node_type="item",
            reaction_type="like",
            target_item_type=target_item_type.value,
            emoji=emoji,
            item_id=item_id,
            reaction_action_source=(
                "double_tap" if reaction_status == ReactionStatus.CREATED else "action_menu"
            ),
            shh_mode=shh_mode,
            item_type=ThreadItemType.REACTION,
            client_context=client_context,
            super_react_type="none",
            send_attribution="direct_thread",
            **kwargs,
        )

    def send_user_story(
        self,
        thread_id: str,
        media_id: str,
        text: str = "",
        shh_mode: bool = False,
        client_context: str | None = None,
    ) -> Awaitable[CommandResponse]:
        return self.send_item(
            thread_id,
            text=text,
            item_id=media_id,
            shh_mode=shh_mode,
            item_type=ThreadItemType.REEL_SHARE,
            client_context=client_context,
        )

    def send_text(
        self,
        thread_id: str,
        text: str = "",
        urls: list[str] | None = None,
        shh_mode: bool = False,
        client_context: str | None = None,
        replied_to_item_id: str | None = None,
        replied_to_client_context: str | None = None,
        mentioned_user_ids: list[int] | None = None,
    ) -> Awaitable[CommandResponse]:
        args = {
            "text": text,
        }
        item_type = ThreadItemType.TEXT
        if urls is not None:
            args = {
                "link_text": text,
                "link_urls": json.dumps(urls or []),
            }
            item_type = ThreadItemType.LINK
        if mentioned_user_ids:
            args["mentioned_user_ids"] = json.dumps([str(x) for x in mentioned_user_ids])
            args["sampled"] = True
        return self.send_item(
            thread_id,
            **args,
            shh_mode=shh_mode,
            item_type=item_type,
            client_context=client_context,
            replied_to_item_id=replied_to_item_id,
            replied_to_client_context=replied_to_client_context,
            send_attribution="direct_thread",
            send_silently=False,
            is_x_transport_forward=False,
        )

    def mark_seen(
        self, thread_id: str, item_id: str, client_context: str | None = None
    ) -> Awaitable[None]:
        return self.send_command(
            thread_id,
            item_id=item_id,
            action=ThreadAction.MARK_SEEN,
            client_context=client_context,
        )

    def mark_visual_item_seen(
        self, thread_id: str, item_id: str, client_context: str | None = None
    ) -> Awaitable[CommandResponse]:
        return self.send_command(
            thread_id,
            item_id=item_id,
            action=ThreadAction.MARK_VISUAL_ITEM_SEEN,
            client_context=client_context,
        )

    def indicate_activity(
        self,
        thread_id: str,
        activity_status: TypingStatus = TypingStatus.TEXT,
        client_context: str | None = None,
    ) -> Awaitable[CommandResponse]:
        return self.send_command(
            thread_id,
            activity_status=activity_status.value,
            action=ThreadAction.INDICATE_ACTIVITY,
            client_context=client_context,
        )

    # endregion