matrix_appservice_kakaotalk/commands/conn.py

@@ -38,22 +38,7 @@ async def set_notice_room(evt: CommandEvent) -> None:
     needs_auth=True,
     management_only=True,
     help_section=SECTION_CONNECTION,
-    help_text="Disconnect from KakaoTalk chats, but remain logged into profile-management commands",
+    help_text="Check if you're logged into KakaoTalk",
-)
-async def disconnect(evt: CommandEvent) -> None:
-    if not evt.sender.is_connected:
-        await evt.reply("You are already disconnected from KakaoTalk chats")
-        return
-    await evt.mark_read()
-    await evt.sender.client.disconnect()
-    await evt.reply("Successfully disconnected from KakaoTalk chats. To reconnect, use the `sync` command.")
-
-
-@command_handler(
-    needs_auth=True,
-    management_only=True,
-    help_section=SECTION_CONNECTION,
-    help_text="Check if you're logged into KakaoTalk & connected to chats",
 )
 async def ping(evt: CommandEvent) -> None:
     if not await evt.sender.is_logged_in():
@@ -62,11 +47,7 @@ async def ping(evt: CommandEvent) -> None:
     await evt.mark_read()
     try:
         own_info = await evt.sender.get_own_info()
-        await evt.reply(
+        await evt.reply(f"You're logged in as {own_info.nickname} (user ID {evt.sender.ktid})")
-            f"You're logged in as {own_info.nickname} (user ID {evt.sender.ktid})."
-             "\n\n"
-            f"You are {'connected to' if evt.sender.is_connected else '**disconnected** from'} KakaoTalk chats.\n\n"
-        )
     except CommandException as e:
         await evt.reply(f"Error from KakaoTalk: {e}")
 
@@ -75,8 +56,8 @@ async def ping(evt: CommandEvent) -> None:
     needs_auth=True,
     management_only=True,
     help_section=SECTION_CONNECTION,
-    help_text="(Re)connect to KakaoTalk chats & sync any missed chat updates",
+    help_text="Resync chats",
-    help_args="[number_of_channels_to_sync]",
+    help_args="[count]",
 )
 async def sync(evt: CommandEvent) -> None:
     try:
@@ -84,7 +65,7 @@ async def sync(evt: CommandEvent) -> None:
     except IndexError:
         sync_count = None
     except ValueError:
-        await evt.reply("**Usage:** `sync [number_of_channels_to_sync]`")
+        await evt.reply("**Usage:** `$cmdprefix+sp logout [--reset-device]`")
         return
 
     await evt.mark_read()

matrix_appservice_kakaotalk/commands/handler.py

@@ -17,7 +17,7 @@ from typing import NamedTuple
 
 from mautrix.bridge.commands import HelpSection
 
-HelpCacheKey = NamedTuple("KTHelpCacheKey", is_management=bool, is_admin=bool, is_logged_in=bool)
+HelpCacheKey = NamedTuple("FBHelpCacheKey", is_management=bool, is_admin=bool, is_logged_in=bool)
 
 SECTION_AUTH = HelpSection("Authentication", 10, "")
 SECTION_CONNECTION = HelpSection("Connection management", 15, "")

matrix_appservice_kakaotalk/config.py

@@ -51,6 +51,9 @@ class Config(BaseBridgeConfig):
 
         copy("homeserver.asmux")
 
+        if self["appservice.bot_avatar"] == "mxc://maunium.net/ddtNPZSKMNqaUzqrHuWvUADv":
+            base["appservice.bot_avatar"] = "mxc://maunium.net/ygtkteZsXnGJLJHRchUwYWak"
+
         copy("appservice.public.enabled")
         copy("appservice.public.prefix")
         copy("appservice.public.external")

matrix_appservice_kakaotalk/example-config.yaml

@@ -213,15 +213,23 @@ bridge:
     # The number of seconds that a disconnection can last without triggering an automatic re-sync
     # and missed message backfilling when reconnecting.
     # Set to 0 to always re-sync, or -1 to never re-sync automatically.
-    resync_max_disconnected_time: 5
+    #resync_max_disconnected_time: 5
     # Should the bridge do a resync on startup?
     sync_on_startup: true
     # Whether or not temporary disconnections should send notices to the notice room.
     # If this is false, disconnections will never send messages and connections will only send
     # messages if it was disconnected for more than resync_max_disconnected_time seconds.
+    # TODO Probably don't need this
     temporary_disconnect_notices: true
     # Disable bridge notices entirely
     disable_bridge_notices: false
+    on_reconnection_fail:
+        # Whether or not the bridge should try to "refresh" the connection if a normal reconnection
+        # attempt fails.
+        refresh: false
+        # Seconds to wait before attempting to refresh the connection, set a list of two items to
+        # to randomize the interval (min, max).
+        wait_for: 0
     # Set this to true to tell the bridge to re-send m.bridge events to all rooms on the next run.
     # This field will automatically be changed back to false after it,
     # except if the config file is not writable.

matrix_appservice_kakaotalk/kt/client/client.py

@@ -42,7 +42,6 @@ from ..types.bson import Long
 from ..types.client.client_session import LoginResult
 from ..types.chat import Chatlog, KnownChatType
 from ..types.oauth import OAuthCredential, OAuthInfo
-from ..types.packet.chat.kickout import KnownKickoutType, KickoutRes
 from ..types.request import (
     deserialize_result,
     ResultType,
@@ -149,12 +148,11 @@ class Client:
     def _oauth_credential(self) -> JSON:
         return self.user.oauth_credential.serialize()
 
-    @property
+    def _get_user_data(self) -> JSON:
-    def _user_data(self) -> JSON:
+        return dict(
-        return {
+            mxid=self.user.mxid,
-            "mxid": self.user.mxid,
+            oauth_credential=self._oauth_credential
-            "oauth_credential": self._oauth_credential,
+        )
-        }
 
     # region HTTP
 
@@ -181,28 +179,19 @@ class Client:
 
     # region post-token commands
 
-    async def start(self) -> ProfileStruct:
+    async def renew(self) -> OAuthInfo:
-        """
+        """Get a new set of tokens from a refresh token."""
-        Initialize user-specific bridging & state by providing a token obtained from a prior login.
+        return await self._api_request_result(OAuthInfo, "renew", oauth_credential=self._oauth_credential)
-        Receive the user's profile info in response.
-        """
-        profile_req_struct = await self._api_user_request_result(ProfileReqStruct, "start")
-        return profile_req_struct.profile
-
-    async def stop(self) -> None:
-        """Immediately stop bridging this user."""
-        self._stop_listen()
-        await self._rpc_client.request("stop", mxid=self.user.mxid)
 
     async def renew_and_save(self) -> None:
         """Renew and save the user's session tokens."""
-        oauth_info = await self._api_request_result(OAuthInfo, "renew", oauth_credential=self._oauth_credential)
+        oauth_info = await self.renew()
         self.user.oauth_credential = oauth_info.credential
         await self.user.save()
 
     async def connect(self) -> LoginResult:
         """
-        Start a new talk session by providing a token obtained from a prior login.
+        Start a new session by providing a token obtained from a prior login.
         Receive a snapshot of account state in response.
         """
         login_result = await self._api_user_request_result(LoginResult, "connect")
@@ -211,12 +200,12 @@ class Client:
         self._start_listen()
         return login_result
 
-    async def disconnect(self) -> None:
+    async def disconnect(self) -> bool:
-        """Disconnect from the talk session, but remain logged in."""
+        connection_existed = await self._rpc_client.request("disconnect", mxid=self.user.mxid)
-        await self._rpc_client.request("disconnect", mxid=self.user.mxid)
+        self._stop_listen()
-        await self._on_disconnect(None)
+        return connection_existed
 
-    async def get_own_profile(self) -> ProfileStruct:
+    async def fetch_logged_in_user(self, post_login: bool = False) -> ProfileStruct:
         profile_req_struct = await self._api_user_request_result(ProfileReqStruct, "get_own_profile")
         return profile_req_struct.profile
 
@@ -300,7 +289,7 @@ class Client:
         renewed = False
         while True:
             try:
-                return await self._api_request_result(result_type, command, **self._user_data, **data)
+                return await self._api_request_result(result_type, command, **self._get_user_data(), **data)
             except InvalidAccessToken:
                 if renewed:
                     raise
@@ -311,7 +300,7 @@ class Client:
         renewed = False
         while True:
             try:
-                return await self._api_request_void(command, **self._user_data, **data)
+                return await self._api_request_void(command, **self._get_user_data(), **data)
             except InvalidAccessToken:
                 if renewed:
                     raise
@@ -327,7 +316,7 @@ class Client:
         await self.user.on_message(
             Chatlog.deserialize(data["chatlog"]),
             Long.deserialize(data["channelId"]),
-            data["channelType"],
+            data["channelType"]
         )
 
     """ TODO
@@ -335,36 +324,16 @@ class Client:
         await self.user.on_receipt(Receipt.deserialize(data["receipt"]))
     """
 
-    async def _on_listen_disconnect(self, data: dict[str, JSON]) -> None:
-        try:
-            res = KickoutRes.deserialize(data)
-        except Exception:
-            self.log.exception("Invalid kickout reason, defaulting to None")
-            res = None
-        await self._on_disconnect(res)
-
-    async def _on_switch_server(self) -> None:
-        # TODO Reconnect automatically instead
-        await self._on_disconnect(KickoutRes(KnownKickoutType.CHANGE_SERVER))
-
-    async def _on_disconnect(self, res: KickoutRes | None) -> None:
-        self._stop_listen()
-        await self.user.on_disconnect(res)
-
 
     def _start_listen(self) -> None:
         # TODO Automate this somehow, like with a fancy enum
-        self._rpc_client.set_event_handlers(self._get_user_cmd("chat"), [self._on_message])
+        self._rpc_client.set_event_handlers(self._get_user_cmd("message"), [self._on_message])
         # TODO many more listeners
-        self._rpc_client.set_event_handlers(self._get_user_cmd("disconnected"), [self._on_listen_disconnect])
-        self._rpc_client.set_event_handlers(self._get_user_cmd("switch_server"), [self._on_switch_server])
 
     def _stop_listen(self) -> None:
         # TODO Automate this somehow, like with a fancy enum
-        self._rpc_client.set_event_handlers(self._get_user_cmd("chat"), [])
+        self._rpc_client.set_event_handlers(self._get_user_cmd("message"), [])
         # TODO many more listeners
-        self._rpc_client.set_event_handlers(self._get_user_cmd("disconnected"), [])
-        self._rpc_client.set_event_handlers(self._get_user_cmd("switch_server"), [])
 
     def _get_user_cmd(self, command) -> str:
         return f"{command}:{self.user.mxid}"

matrix_appservice_kakaotalk/kt/client/errors.py

@@ -17,6 +17,8 @@
 
 from __future__ import annotations
 
+from typing import Type
+
 from ..types.api.auth_api_client import KnownAuthStatusCode
 from ..types.request import KnownDataStatusCode, RootCommandResult
 

matrix_appservice_kakaotalk/kt/types/chat/attachment/__init__.py

@@ -35,8 +35,8 @@ class PathAttachment(Attachment):
     s: int
 
 
+from .media import *
 from .reply import *
-from .mention import *
 from .media import *
 from .emoticon import *
 #from .voip import *

matrix_appservice_kakaotalk/kt/types/chat/chat.py

@@ -68,8 +68,6 @@ _attachment_type_map: dict[KnownChatType, Type[Attachment]] = {
     KnownChatType.CONTACT: ContactAttachment,
     KnownChatType.AUDIO: AudioAttachment,
     KnownChatType.DITEMEMOTICON: EmoticonAttachment,
-    KnownChatType.STICKER: EmoticonAttachment,
-    KnownChatType.STICKERANI: EmoticonAttachment,
     KnownChatType.SCHEDULE: OpenScheduleAttachment,
     KnownChatType.VOTE: VoteAttachment,
     KnownChatType.MAP: MapAttachment,

matrix_appservice_kakaotalk/kt/types/packet/chat/kickout.py

@@ -1,41 +0,0 @@
-# matrix-appservice-kakaotalk - A Matrix-KakaoTalk puppeting bridge.
-# Copyright (C) 2022 Tulir Asokan, Andrew Ferrazzutti
-#
-# 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 typing import Union
-from enum import IntEnum
-
-from attr import dataclass
-
-from mautrix.types import SerializableAttrs
-
-
-@dataclass
-class KnownKickoutType(IntEnum):
-    CHANGE_SERVER = -2
-    LOGIN_ANOTHER = 0
-    ACCOUNT_DELETED = 1
-
-KickoutType = Union[KnownKickoutType, int]
-
-@dataclass
-class KickoutRes(SerializableAttrs):
-    reason: KickoutType
-
-
-__all__ = [
-    "KnownKickoutType",
-    "KickoutType",
-    "KickoutRes",
-]

matrix_appservice_kakaotalk/matrix.py

@@ -16,6 +16,7 @@
 from __future__ import annotations
 
 from typing import TYPE_CHECKING, Union
+import time
 
 from mautrix.bridge import BaseMatrixHandler
 from mautrix.errors import MatrixError
@@ -38,6 +39,7 @@ from mautrix.types import (
 )
 
 from . import portal as po, puppet as pu, user as u
+from .db import Message as DBMessage
 
 if TYPE_CHECKING:
     from .__main__ import KakaoTalkBridge
@@ -51,10 +53,6 @@ class MatrixHandler(BaseMatrixHandler):
         self.user_id_suffix = f"{suffix}:{homeserver}"
         super().__init__(bridge=bridge)
 
-    @staticmethod
-    async def allow_bridging_message(user: u.User, portal: po.Portal) -> bool:
-        return user.is_connected or (user.relay_whitelisted and portal.has_relay)
-
     async def send_welcome_message(self, room_id: RoomID, inviter: u.User) -> None:
         await super().send_welcome_message(room_id, inviter)
         if not inviter.notice_room:
@@ -236,7 +234,7 @@ class MatrixHandler(BaseMatrixHandler):
         event_id: EventID,
         data: SingleReceiptEventContent,
     ) -> None:
-        self.log.info("TODO: handle_read_receipt")
+        self.log.info("TODO")
         """
         if not user.mqtt:
             return

matrix_appservice_kakaotalk/portal.py

@@ -1260,10 +1260,7 @@ class Portal(DBPortal, BasePortal):
             # TODO Save kt_sender in DB instead? Depends on if DM channels are shared...
             user = await u.User.get_by_ktid(self.kt_receiver)
             assert user, f"Found no user for this portal's receiver of {self.kt_receiver}"
-            if user.is_connected:
+            await self._update_participants(user)
-                await self._update_participants(user)
-            else:
-                self.log.debug(f"Not setting _main_intent of new direct chat for disconnected user {user.ktid}")
         else:
             self.log.debug("Not setting _main_intent of new direct chat until after checking participant list")
 

matrix_appservice_kakaotalk/rpc/rpc.py

@@ -63,7 +63,7 @@ class RPCClient:
                     r, w = await asyncio.open_unix_connection(self.config["rpc.connection.path"])
                     break
                 except:
-                    self.log.warning(f'No unix socket available at {self.config["rpc.connection.path"]}, wait for it to exist...')
+                    self.log.warn(f'No unix socket available at {self.config["rpc.connection.path"]}, wait for it to exist...')
                     await asyncio.sleep(10)
         elif self.config["rpc.connection.type"] == "tcp":
             while True:
@@ -72,7 +72,7 @@ class RPCClient:
                                                          self.config["rpc.connection.port"])
                     break
                 except:
-                    self.log.warning(f'No TCP connection open at {self.config["rpc.connection.host"]}:{self.config["rpc.connection.path"]}, wait for it to become available...')
+                    self.log.warn(f'No TCP connection open at {self.config["rpc.connection.host"]}:{self.config["rpc.connection.path"]}, wait for it to become available...')
                     await asyncio.sleep(10)
         else:
             raise RuntimeError("invalid rpc connection type")

matrix_appservice_kakaotalk/user.py

@@ -45,9 +45,17 @@ from .kt.types.chat.chat import Chatlog
 from .kt.types.client.client_session import LoginDataItem, LoginResult
 from .kt.types.oauth import OAuthCredential
 from .kt.types.openlink.open_channel_info import OpenChannelData, OpenChannelInfo
-from .kt.types.packet.chat.kickout import KnownKickoutType, KickoutRes
 
-METRIC_CONNECT_AND_SYNC = Summary("bridge_connect_and_sync", "calls to connect_and_sync")
+METRIC_CONNECT_AND_SYNC = Summary("bridge_sync_channels", "calls to connect_and_sync")
+METRIC_MEMBERS_ADDED = Summary("bridge_on_members_added", "calls to on_members_added")
+METRIC_MEMBER_REMOVED = Summary("bridge_on_member_removed", "calls to on_member_removed")
+METRIC_TYPING = Summary("bridge_on_typing", "calls to on_typing")
+METRIC_PRESENCE = Summary("bridge_on_presence", "calls to on_presence")
+METRIC_REACTION = Summary("bridge_on_reaction", "calls to on_reaction")
+METRIC_MESSAGE_UNSENT = Summary("bridge_on_unsent", "calls to on_unsent")
+METRIC_MESSAGE_SEEN = Summary("bridge_on_message_seen", "calls to on_message_seen")
+METRIC_TITLE_CHANGE = Summary("bridge_on_title_change", "calls to on_title_change")
+METRIC_AVATAR_CHANGE = Summary("bridge_on_avatar_change", "calls to on_avatar_change")
 METRIC_MESSAGE = Summary("bridge_on_message", "calls to on_message")
 METRIC_LOGGED_IN = Gauge("bridge_logged_in", "Users logged into the bridge")
 METRIC_CONNECTED = Gauge("bridge_connected", "Bridge users connected to KakaoTalk")
@@ -58,16 +66,15 @@ if TYPE_CHECKING:
 BridgeState.human_readable_errors.update(
     {
         "kt-reconnection-error": "Failed to reconnect to KakaoTalk",
-        # TODO Use for errors in Node backend that cause session to be lost
+        "kt-connection-error": "KakaoTalk disconnected unexpectedly",
-        #"kt-connection-error": "KakaoTalk disconnected unexpectedly",
         "kt-auth-error": "Authentication error from KakaoTalk: {message}",
+        "kt-disconnected": None,
         "logged-out": "You're not logged into KakaoTalk",
     }
 )
 
 
 class User(DBUser, BaseUser):
-    temp_disconnect_notices: bool = True
     shutdown: bool = False
     config: Config
 
@@ -131,7 +138,7 @@ class User(DBUser, BaseUser):
         cls.bridge = bridge
         cls.config = bridge.config
         cls.az = bridge.az
-        cls.temp_disconnect_notices = bridge.config["bridge.temporary_disconnect_notices"]
+        cls.loop = bridge.loop
         return (user.reload_session(is_startup=True) async for user in cls.all_logged_in())
 
     @property
@@ -238,17 +245,17 @@ class User(DBUser, BaseUser):
         self.access_token = oauth_credential.accessToken
         self.refresh_token = oauth_credential.refreshToken
         if self.uuid != oauth_credential.deviceUUID:
-            self.log.warning(f"UUID mismatch: expected {self.uuid}, got {oauth_credential.deviceUUID}")
+            self.log.warn(f"UUID mismatch: expected {self.uuid}, got {oauth_credential.deviceUUID}")
             self.uuid = oauth_credential.deviceUUID
 
     async def get_own_info(self) -> ProfileStruct:
         if not self._logged_in_info or self._logged_in_info_time + 60 * 60 < time.monotonic():
-            self._logged_in_info = await self.client.get_own_profile()
+            self._logged_in_info = await self.client.fetch_logged_in_user()
             self._logged_in_info_time = time.monotonic()
         return self._logged_in_info
 
     async def _load_session(self, is_startup: bool) -> bool:
-        if self._is_logged_in and is_startup:
+        if self._is_logged_in and not is_startup:
             return True
         elif not self.has_state:
             # If we have a user in the DB with no state, we can assume
@@ -259,17 +266,18 @@ class User(DBUser, BaseUser):
             )
             return False
         client = Client(self, log=self.log.getChild("ktclient"))
-        user_info = await client.start()
+        user_info = await self.fetch_logged_in_user(client)
-        # NOTE On failure, client.start throws instead of returning False
+        if user_info:
-        self.log.info("Loaded session successfully")
+            self.log.info("Loaded session successfully")
-        self.client = client
+            self.client = client
-        self._logged_in_info = user_info
+            self._logged_in_info = user_info
-        self._logged_in_info_time = time.monotonic()
+            self._logged_in_info_time = time.monotonic()
-        self._track_metric(METRIC_LOGGED_IN, True)
+            self._track_metric(METRIC_LOGGED_IN, True)
-        self._is_logged_in = True
+            self._is_logged_in = True
-        self.is_connected = None
+            self.is_connected = None
-        asyncio.create_task(self.post_login(is_startup=is_startup))
+            asyncio.create_task(self.post_login(is_startup=is_startup))
-        return True
+            return True
+        return False
 
     async def _send_reset_notice(self, e: AuthenticationRequired, edit: EventID | None = None) -> None:
         await self.send_bridge_notice(
@@ -285,6 +293,22 @@ class User(DBUser, BaseUser):
         )
         await self.logout(remove_ktid=False)
 
+    async def fetch_logged_in_user(
+        self, client: Client | None = None, action: str = "restore session"
+    ) -> ProfileStruct:
+        if not client:
+            client = self.client
+        # TODO Retry network connection failures here, or in the client (like token refreshes are)?
+        try:
+            return await client.fetch_logged_in_user()
+        except AuthenticationRequired as e:
+            if action != "restore session":
+                await self._send_reset_notice(e)
+            raise
+        except Exception:
+            self.log.exception(f"Failed to {action}")
+            raise
+
     async def is_logged_in(self, _override: bool = False) -> bool:
         if not self.has_state or not self.client:
             return False
@@ -336,7 +360,11 @@ class User(DBUser, BaseUser):
     async def logout(self, *, remove_ktid: bool = True, reset_device: bool = False) -> None:
         if self.client:
             # TODO Look for a logout API call
-            await self.client.stop()
+            was_connected = await self.client.disconnect()
+            if was_connected != self._is_connected:
+                self.log.warn(
+                    f"Node backend was{' not' if not was_connected else ''} connected, "
+                    f"but we thought it was{' not' if not self._is_connected else ''}")
         if remove_ktid:
             await self.push_bridge_state(BridgeStateEvent.LOGGED_OUT)
         self._track_metric(METRIC_LOGGED_IN, False)
@@ -375,7 +403,6 @@ class User(DBUser, BaseUser):
             sync_count = self.config["bridge.initial_chat_sync"]
         else:
             sync_count = None
-        # TODO Don't auto-connect on startup if user's last state was disconnected
         await self.connect_and_sync(sync_count)
 
     async def get_direct_chats(self) -> dict[UserID, list[RoomID]]:
@@ -390,7 +417,7 @@ class User(DBUser, BaseUser):
         # TODO Look for a way to sync all channels without (re-)logging in
         try:
             login_result = await self.client.connect()
-            await self.on_connect()
+            await self.push_bridge_state(BridgeStateEvent.CONNECTED)
             await self._sync_channels(login_result, sync_count)
             return True
         except AuthenticationRequired as e:
@@ -545,9 +572,6 @@ class User(DBUser, BaseUser):
         state = BridgeState(state_event=BridgeStateEvent.UNKNOWN_ERROR)
         if self.is_connected:
             state.state_event = BridgeStateEvent.CONNECTED
-        # TODO
-        #elif self._is_logged_in and self._is_reconnecting:
-        #    state.state_event = BridgeStateEvent.TRANSIENT_DISCONNECT
         return [state]
 
     async def get_puppet(self) -> pu.Puppet | None:
@@ -557,43 +581,6 @@ class User(DBUser, BaseUser):
 
     # region KakaoTalk event handling
 
-    async def on_connect(self) -> None:
-        now = time.monotonic()
-        disconnected_at = self._connection_time
-        max_delay = self.config["bridge.resync_max_disconnected_time"]
-        first_connect = self.is_connected is None
-        self.is_connected = True
-        self._track_metric(METRIC_CONNECTED, True)
-        if not first_connect and disconnected_at + max_delay < now:
-            duration = int(now - disconnected_at)
-            self.log.debug(f"Disconnection lasted {duration} seconds")
-        elif self.temp_disconnect_notices:
-            await self.send_bridge_notice("Connected to KakaoTalk chats")
-        await self.push_bridge_state(BridgeStateEvent.CONNECTED)
-
-    async def on_disconnect(self, res: KickoutRes | None) -> None:
-        self.is_connected = False
-        self._track_metric(METRIC_CONNECTED, False)
-        if res:
-            logout = False
-            if res.reason == KnownKickoutType.LOGIN_ANOTHER:
-                reason_str = "Logged in from another desktop client."
-            elif res.reason == KnownKickoutType.CHANGE_SERVER:
-                # TODO Reconnect automatically instead
-                reason_str = "KakaoTalk backend server changed."
-            elif res.reason == KnownKickoutType.ACCOUNT_DELETED:
-                reason_str = "Your KakaoTalk account was deleted!"
-                logout = True
-            else:
-                reason_str = f"Unknown reason ({res.reason})."
-            if not logout:
-                reason_suffix = " To reconnect, use the `sync` command."
-                # TODO What bridge state to push?
-            else:
-                reason_suffix = " You are now logged out."
-                await self.logout()
-            await self.send_bridge_notice(f"Disconnected from KakaoTalk: {reason_str} {reason_suffix}")
-
     async def on_logged_in(self, oauth_credential: OAuthCredential) -> None:
         self.log.debug(f"Successfully logged in as {oauth_credential.userId}")
         self.oauth_credential = oauth_credential
@@ -601,10 +588,11 @@ class User(DBUser, BaseUser):
         self.client = Client(self, log=self.log.getChild("ktclient"))
         await self.save()
         self._is_logged_in = True
-        # TODO Retry network connection failures here, or in the client (like token refreshes are)?
+        try:
-        #      Should also catch unlikely authentication errors
+            self._logged_in_info = await self.client.fetch_logged_in_user(post_login=True)
-        self._logged_in_info = await self.client.start()
+            self._logged_in_info_time = time.monotonic()
-        self._logged_in_info_time = time.monotonic()
+        except Exception:
+            self.log.exception("Failed to fetch post-login info")
         asyncio.create_task(self.post_login(is_startup=True))
 
     @async_time(METRIC_MESSAGE)

node/src/client.js

@@ -63,40 +63,30 @@ class UserClient {
 	/**
 	 * DO NOT CONSTRUCT DIRECTLY. Callers should use {@link UserClient#create} instead.
 	 * @param {string} mxid
-	 * @param {PeerClient} peerClient TODO Make RPC user-specific instead of needing this
+	 * @param {OAuthCredential} credential
 	 */
-	constructor(mxid, peerClient) {
+	constructor(mxid, credential) {
 		if (!UserClient.#initializing) {
 			throw new Error("Private constructor")
 		}
 		UserClient.#initializing = false
 
 		this.mxid = mxid
-		this.peerClient = peerClient
+		this.credential = credential
 	}
 
 	/**
 	 * @param {string} mxid The ID of the associated Matrix user
-	 * @param {OAuthCredential} credential The token to log in with, obtained from prior login
+	 * @param {OAuthCredential} credential The tokens that API calls may use
-	 * @param {PeerClient} peerClient What handles RPC
 	 */
-	static async create(mxid, credential, peerClient) {
+	static async create(mxid, credential) {
 		this.#initializing = true
-		const userClient = new UserClient(mxid, peerClient)
+		const userClient = new UserClient(mxid, credential)
 
 		userClient.#serviceClient = await ServiceApiClient.create(credential)
 		return userClient
 	}
 
-
-	log(...text) {
-		console.log(`[API/${this.mxid}]`, ...text)
-	}
-
-	error(...text) {
-		console.error(`[API/${this.mxid}]`, ...text)
-	}
-
 	/**
 	 * @param {Object} channel_props
 	 * @param {Long} channel_props.id
@@ -121,68 +111,16 @@ class UserClient {
 		}
 	}
 
-	/**
+	close() {
-	 * @param {OAuthCredential} credential The token to log in with, obtained from prior login
+		this.#talkClient.close()
-	 */
-	async connect(credential) {
-		// TODO Don't re-login if possible. But must still return a LoginResult!
-		this.disconnect()
-		const res = await this.#talkClient.login(credential)
-		if (!res.success) return res
-
-		this.#talkClient.on("chat", (data, channel) => {
-			this.log(`Received chat message ${data.chat.logId} in channel ${channel.channelId}`)
-			return this.write("chat", {
-				//is_sequential: true, // TODO Make sequential per user & channel (if it isn't already)
-				chatlog: data.chat,
-				channelId: channel.channelId,
-				channelType: channel.info.type,
-			})
-		})
-
-		/* TODO Many more listeners
-		this.#talkClient.on("chat_read", (chat, channel, reader) => {
-			this.log(`chat_read in channel ${channel.channelId}`)
-			//chat.logId
-		})
-		*/
-
-		this.#talkClient.on("disconnected", (reason) => {
-			this.log(`Disconnected (reason=${reason})`)
-			this.disconnect()
-			return this.write("disconnected", {
-				reason: reason,
-			})
-		})
-
-		this.#talkClient.on("switch_server", () => {
-			this.log(`Server switch requested`)
-			return this.write("switch_server", {
-				is_sequential: true,
-			})
-		})
-
-		return res
-	}
-
-	disconnect() {
-		if (this.#talkClient.logon) {
-			this.#talkClient.close()
-		}
 	}
 
 	/**
-	 * Send a user-specific command with (optional) data to the socket.
+	 * TODO Maybe use a "write" method instead
-	 *
+	 * @param {string} command
-	 * @param {string} command - The data to write.
-	 * @param {?object} data - The data to write.
 	 */
-	write(command, data) {
+	getCmd(command) {
-		return this.peerClient.write({
+		return `${command}:${this.mxid}`
-			id: --this.peerClient.notificationID,
-			command: `${command}:${this.mxid}`,
-			...data
-		})
 	}
 }
 
@@ -245,7 +183,7 @@ export default class PeerClient {
 		this.stopped = true
 		this.#closeUsers()
 		try {
-			await this.write({ id: --this.notificationID, command: "quit", error })
+			await this.#write({ id: --this.notificationID, command: "quit", error })
 			await promisify(cb => this.socket.end(cb))
 		} catch (err) {
 			this.error("Failed to end connection:", err)
@@ -265,7 +203,7 @@ export default class PeerClient {
 	#closeUsers() {
 		this.log("Closing all API clients for", this.peerID)
 		for (const userClient of this.userClients.values()) {
-			userClient.disconnect()
+			userClient.close()
 		}
 		this.userClients.clear()
 	}
@@ -276,7 +214,7 @@ export default class PeerClient {
 	 * @param {object} data - The data to write.
 	 * @returns {Promise<void>}
 	 */
-	write(data) {
+	#write(data) {
 		return promisify(cb => this.socket.write(JSON.stringify(data, this.#writeReplacer) + "\n", cb))
 	}
 
@@ -293,7 +231,7 @@ export default class PeerClient {
 	}
 
 	/**
-	 * Obtain login tokens. If this fails due to not having a device, also request a device passcode.
+	 * Log in. If this fails due to not having a device, also request a device passcode.
 	 * @param {Object} req
 	 * @param {string} req.uuid
 	 * @param {Object} req.form
@@ -346,12 +284,22 @@ export default class PeerClient {
 		return this.userClients.get(mxid)
 	}
 
+	/**
+	 * Get the service client for the specified user ID, or create
+	 * and return a new service client if no user ID is provided.
+	 * @param {?string} mxid
+	 * @param {?OAuthCredential} oauth_credential
+	 */
+	async #getServiceClient(mxid, oauth_credential) {
+		return this.#tryGetUser(mxid)?.serviceClient ||
+			await ServiceApiClient.create(oauth_credential)
+	}
+	
 	/**
 	 * @param {Object} req
 	 * @param {OAuthCredential} req.oauth_credential
 	 */
 	handleRenew = async (req) => {
-		// TODO Cache per user? Reset API client objects?
 		const oAuthClient = await OAuthApiClient.create()
 		return await oAuthClient.renew(req.oauth_credential)
 	}
@@ -361,57 +309,72 @@ export default class PeerClient {
 	 * @param {string} req.mxid
 	 * @param {OAuthCredential} req.oauth_credential
 	 */
-	userStart = async (req) => {
+	handleConnect = async (req) => {
-		const userClient = this.#tryGetUser(req.mxid) || await UserClient.create(req.mxid, req.oauth_credential, this)
+		// TODO Don't re-login if possible. But must still return a LoginResult!
-		// TODO Should call requestMore/LessSettings instead
+		this.handleDisconnect(req)
-		const res = await userClient.serviceClient.requestMyProfile()
+
-		if (res.success) {
+		const userClient = await UserClient.create(req.mxid, req.oauth_credential)
-			this.userClients.set(req.mxid, userClient)
+		const res = await userClient.talkClient.login(req.oauth_credential)
-		}
+		if (!res.success) return res
+
+		this.userClients.set(req.mxid, userClient)
+
+		userClient.talkClient.on("chat", (data, channel) => {
+			this.log(`Received message ${data.chat.logId} in channel ${channel.channelId}`)
+			return this.#write({
+				id: --this.notificationID,
+				command: userClient.getCmd("message"),
+				//is_sequential: true, // TODO Make sequential per user & channel (if it isn't already)
+				chatlog: data.chat,
+				channelId: channel.channelId,
+				channelType: channel.info.type,
+			})
+		})
+
+		/* TODO Many more listeners
+		userClient.talkClient.on("chat_read", (chat, channel, reader) => {
+			this.log(`chat_read in channel ${channel.channelId}`)
+			//chat.logId
+		})
+		*/
+
 		return res
 	}
 
-	/**
-	 * @param {Object} req
-	 * @param {string} req.mxid
-	 */
-	userStop = async (req) => {
-		this.handleDisconnect(req)
-		this.userClients.delete(req.mxid)
-	}
-
-	/**
-	 * @param {Object} req
-	 * @param {string} req.mxid
-	 * @param {OAuthCredential} req.oauth_credential
-	 */
-	handleConnect = async (req) => {
-		return await this.#getUser(req.mxid).connect(req.oauth_credential)
-	}
-
 	/**
 	 * @param {Object} req
 	 * @param {string} req.mxid
 	 */
 	handleDisconnect = (req) => {
-		this.#tryGetUser(req.mxid)?.disconnect()
+		const userClient = this.#tryGetUser(req.mxid)
+		if (!!userClient) {
+			userClient.close()
+			this.userClients.delete(req.mxid)
+			return true
+		} else {
+			return false
+		}
 	}
 
 	/**
 	 * @param {Object} req
 	 * @param {?string} req.mxid
+	 * @param {?OAuthCredential} req.oauth_credential
 	 */
 	getOwnProfile = async (req) => {
-		return await this.#getUser(req.mxid).serviceClient.requestMyProfile()
+		const serviceClient = await this.#getServiceClient(req.mxid, req.oauth_credential)
+		return await serviceClient.requestMyProfile()
 	}
 
 	/**
 	 * @param {Object} req
 	 * @param {?string} req.mxid
+	 * @param {?OAuthCredential} req.oauth_credential
 	 * @param {Long} req.user_id
 	 */
 	getProfile = async (req) => {
-		return await this.#getUser(req.mxid).serviceClient.requestProfile(req.user_id)
+		const serviceClient = await this.#getServiceClient(req.mxid, req.oauth_credential)
+		return await serviceClient.requestProfile(req.user_id)
 	}
 
 	/**
@@ -468,7 +431,8 @@ export default class PeerClient {
 	 * @param {?OAuthCredential} req.oauth_credential
 	 */
 	listFriends = async (req) => {
-		return await this.#getUser(req.mxid).serviceClient.requestFriendList()
+		const serviceClient = await this.#getServiceClient(req.mxid, req.oauth_credential)
+		return await serviceClient.requestFriendList()
 	}
 
 	/**
@@ -580,8 +544,6 @@ export default class PeerClient {
 				register_device: this.registerDevice,
 				login: this.handleLogin,
 				renew: this.handleRenew,
-				start: this.userStart,
-				stop: this.userStop,
 				connect: this.handleConnect,
 				disconnect: this.handleDisconnect,
 				get_own_profile: this.getOwnProfile,
@@ -613,7 +575,7 @@ export default class PeerClient {
 				// TODO Check if session is broken. If it is, close the PeerClient
 			}
 		}
-		await this.write(resp)
+		await this.#write(resp)
 	}
 
 	#writeReplacer = function(key, value) {