Inbound replies (and some kwarg cleanup)

matrix_appservice_kakaotalk/db/message.py

@@ -110,9 +110,9 @@ class Message:
         event_ids: list[EventID],
         timestamp: int,
         mx_room: RoomID,
-    ) -> None:
+    ) -> list[Message]:
         if not event_ids:
-            return
+            return []
         columns = [col.strip('"') for col in cls.columns.split(", ")]
         records = [
             (mxid, mx_room, ktid, index, kt_chat, kt_receiver, timestamp)
@@ -123,6 +123,7 @@ class Message:
                 await conn.copy_records_to_table("message", records=records, columns=columns)
             else:
                 await conn.executemany(cls._insert_query, records)
+        return [Message(*record) for record in records]
 
 
     async def insert(self) -> None:

matrix_appservice_kakaotalk/db/reaction.py

@@ -43,6 +43,16 @@ class Reaction:
             return None
         return cls(**row)
 
+    @classmethod
+    async def get_by_message_ktid(cls, kt_msgid: str, kt_receiver: int) -> dict[int, Reaction]:
+        q = (
+            "SELECT mxid, mx_room, kt_msgid, kt_receiver, kt_sender, reaction "
+            "FROM reaction WHERE kt_msgid=$1 AND kt_receiver=$2"
+        )
+        rows = await cls.db.fetch(q, kt_msgid, kt_receiver)
+        row_gen = (cls._from_row(row) for row in rows)
+        return {react.kt_sender: react for react in row_gen}
+
     @classmethod
     async def get_by_ktid(cls, kt_msgid: str, kt_receiver: int, kt_sender: int) -> Reaction | None:
         q = (

matrix_appservice_kakaotalk/example-config.yaml

@@ -19,6 +19,9 @@ homeserver:
     status_endpoint: null
     # Endpoint for reporting per-message status.
     message_send_checkpoint_endpoint: null
+    # Whether asynchronous uploads via MSC2246 should be enabled for media.
+    # Requires a media repo that supports MSC2246.
+    async_media: false
 
 # Application service host/registration related details
 # Changing these values requires regeneration of the registration.

matrix_appservice_kakaotalk/kt/client/client.py

@@ -53,7 +53,7 @@ from ..types.request import (
 
 from .types import PortalChannelInfo, UserInfoUnion, ChannelProps
 
-from .errors import InvalidAccessToken
+from .errors import InvalidAccessToken, CommandException
 from .error_helper import raise_unsuccessful_response
 
 try:
@@ -257,10 +257,26 @@ class Client:
             "list_friends",
         )
 
+    async def get_friend_dm_id(self, friend_id: Long) -> Long | None:
+        try:
+            return await self._api_user_request_result(
+                Long,
+                "get_friend_dm_id",
+                friend_id=friend_id.serialize(),
+            )
+        except CommandException:
+            self.log.exception(f"Could not find friend with ID {friend_id}")
+            return None
+
+    async def get_memo_ids(self) -> list[Long]:
+        return ResultListType(Long).deserialize(
+            await self._rpc_client.request("get_memo_ids", mxid=self.user.mxid)
+        )
+
     async def send_message(self, channel_props: ChannelProps, text: str) -> Chatlog:
         return await self._api_user_request_result(
             Chatlog,
-            "send_message",
+            "send_chat",
             channel_props=channel_props.serialize(),
             text=text,
         )

matrix_appservice_kakaotalk/kt/types/user/channel_user_info.py

@@ -13,6 +13,8 @@
 #
 # 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 Optional
+
 from attr import dataclass
 
 from ..openlink.open_link_type import OpenChannelUserPerm
@@ -36,12 +38,12 @@ class ChannelUserInfo(DisplayUserInfo):
 
 @dataclass
 class NormalChannelUserInfo(ChannelUserInfo):
-    countryIso: str
     accountId: int
     statusMessage: str
     linkedServices: str
     ut: int
-    suspended: bool
+    countryIso: Optional[str] = None # NOTE Made optional
+    suspended: Optional[bool] = None # NOTE Made optional
 
 
 @dataclass

matrix_appservice_kakaotalk/matrix.py

@@ -64,80 +64,6 @@ class MatrixHandler(BaseMatrixHandler):
                 room_id, "This room has been marked as your KakaoTalk bridge notice room."
             )
 
-    async def handle_puppet_invite(
-        self, room_id: RoomID, puppet: pu.Puppet, invited_by: u.User, event_id: EventID
-    ) -> None:
-        intent = puppet.default_mxid_intent
-        self.log.debug(f"{invited_by.mxid} invited puppet for {puppet.ktid} to {room_id}")
-        if not await invited_by.is_logged_in():
-            await intent.error_and_leave(
-                room_id,
-                text="Please log in before inviting KakaoTalk puppets to private chats.",
-            )
-            return
-
-        portal = await po.Portal.get_by_mxid(room_id)
-        if portal:
-            if portal.is_direct:
-                await intent.error_and_leave(
-                    room_id, text="You can not invite additional users to private chats."
-                )
-                return
-            # TODO add KakaoTalk inviting
-            # await portal.invite_kakaotalk(inviter, puppet)
-            # await intent.join_room(room_id)
-            return
-        await intent.join_room(room_id)
-        try:
-            members = await intent.get_room_members(room_id)
-        except MatrixError:
-            self.log.exception(f"Failed to get member list after joining {room_id}")
-            await intent.leave_room(room_id)
-            return
-        if len(members) > 2:
-            # TODO add KakaoTalk group creating
-            await intent.send_notice(
-                room_id, "You can not invite KakaoTalk puppets to multi-user rooms."
-            )
-            await intent.leave_room(room_id)
-            return
-        portal = await po.Portal.get_by_ktid(
-            puppet.ktid, fb_receiver=invited_by.ktid # TODO kt_type=??
-        )
-        if portal.mxid:
-            try:
-                await intent.invite_user(portal.mxid, invited_by.mxid, check_cache=False)
-                await intent.send_notice(
-                    room_id,
-                    text=f"You already have a private chat with me in room {portal.mxid}",
-                    html=(
-                        "You already have a private chat with me: "
-                        f"<a href='https://matrix.to/#/{portal.mxid}'>Link to room</a>"
-                    ),
-                )
-                await intent.leave_room(room_id)
-                return
-            except MatrixError:
-                pass
-        portal.mxid = room_id
-        e2be_ok = await portal.check_dm_encryption()
-        await portal.save()
-        if e2be_ok is True:
-            evt_type, content = await self.e2ee.encrypt(
-                room_id,
-                EventType.ROOM_MESSAGE,
-                TextMessageEventContent(
-                    msgtype=MessageType.NOTICE,
-                    body="Portal to private chat created and end-to-bridge encryption enabled.",
-                ),
-            )
-            await intent.send_message_event(room_id, evt_type, content)
-        else:
-            message = "Portal to private chat created."
-            if e2be_ok is False:
-                message += "\n\nWarning: Failed to enable end-to-bridge encryption"
-            await intent.send_notice(room_id, message)
-
     async def handle_invite(
         self, room_id: RoomID, user_id: UserID, invited_by: u.User, event_id: EventID
     ) -> None:

matrix_appservice_kakaotalk/portal.py

@@ -20,7 +20,6 @@ from typing import (
     Any,
     AsyncGenerator,
     Awaitable,
-    Callable,
     Pattern,
     cast,
 )
@@ -32,7 +31,7 @@ import time
 
 from mautrix.appservice import IntentAPI
 from mautrix.bridge import BasePortal, NotificationDisabler, async_getter_lock
-from mautrix.errors import MatrixError
+from mautrix.errors import MatrixError, MForbidden, MNotFound, SessionNotFound
 from mautrix.types import (
     AudioInfo,
     ContentURI,
@@ -67,14 +66,14 @@ from .formatter import kakaotalk_to_matrix, matrix_to_kakaotalk
 from .kt.types.bson import Long
 from .kt.types.channel.channel_info import ChannelInfo
 from .kt.types.channel.channel_type import KnownChannelType, ChannelType
-from .kt.types.chat import Chatlog, KnownChatType
+from .kt.types.chat import Chatlog, ChatType, KnownChatType
 from .kt.types.chat.attachment import (
-    Attachment,
     AudioAttachment,
     #FileAttachment,
     MediaAttachment,
     MultiPhotoAttachment,
     PhotoAttachment,
+    ReplyAttachment,
     VideoAttachment,
 )
 
@@ -122,7 +121,7 @@ class Portal(DBPortal, BasePortal):
     config: Config
 
     _main_intent: IntentAPI | None
-    _kt_sender: int | None
+    _kt_sender: Long | None
     _create_room_lock: asyncio.Lock
     _send_locks: dict[int, asyncio.Lock]
     _noop_lock: FakeLock = FakeLock()
@@ -190,21 +189,9 @@ class Portal(DBPortal, BasePortal):
         NotificationDisabler.config_enabled = cls.config["bridge.backfill.disable_notifications"]
 
         # TODO More
-        cls._message_type_handler_map: dict[
-            KnownChatType,
-            Callable[
-                [
-                    Portal,
-                    u.User,
-                    IntentAPI,
-                    Attachment | None,
-                    int,
-                    str | None
-                ],
-                Awaitable[list[EventID]]
-            ]
-        ] = {
+        cls._message_type_handler_map = {
             KnownChatType.TEXT: cls._handle_remote_text,
+            KnownChatType.REPLY: cls._handle_remote_reply,
             KnownChatType.PHOTO: cls._handle_remote_photo,
             KnownChatType.MULTIPHOTO: cls._handle_remote_multiphoto,
             KnownChatType.VIDEO: cls._handle_remote_video,
@@ -217,15 +204,20 @@ class Portal(DBPortal, BasePortal):
     async def delete(self) -> None:
         if self.mxid:
             await DBMessage.delete_all_by_room(self.mxid)
-        self.by_ktid.pop(self._ktid_full, None)
-        self.by_mxid.pop(self.mxid, None)
-        await super().delete()
+            self.by_mxid.pop(self.mxid, None)
+        self.by_ktid.pop(self.ktid_full, None)
+        self.mxid = None
+        self.name_set = False
+        self.avatar_set = False
+        self.relay_user_id = None
+        self.encrypted = False
+        await super().save()
 
     # endregion
     # region Properties
 
     @property
-    def _ktid_full(self) -> tuple[int, int]:
+    def ktid_full(self) -> tuple[Long, Long]:
         return self.ktid, self.kt_receiver
 
     @property
@@ -258,13 +250,14 @@ class Portal(DBPortal, BasePortal):
     @property
     def main_intent(self) -> IntentAPI:
         if not self._main_intent:
-            raise ValueError(
-                "Portal must be postinit()ed before main_intent can be used"
-                if not self.is_direct else
-                "Direct chat portal must call postinit and _update_participants before main_intent can be used"
-            )
+            raise ValueError("Portal must be postinit()ed before main_intent can be used")
         return self._main_intent
 
+    async def get_dm_puppet(self) -> p.Puppet | None:
+        if not self.is_direct:
+            return None
+        return await p.Puppet.get_by_ktid(self.kt_sender)
+
     # endregion
     # region Chat info updating
 
@@ -362,7 +355,12 @@ class Portal(DBPortal, BasePortal):
             data, decryption_info = encrypt_attachment(data)
             upload_mime_type = "application/octet-stream"
             filename = None
-        url = await intent.upload_media(data, mime_type=upload_mime_type, filename=filename)
+        url = await intent.upload_media(
+            data,
+            mime_type=upload_mime_type,
+            filename=filename,
+            async_upload=cls.config["homeserver.async_media"],
+        )
         if decryption_info:
             decryption_info.url = url
         return url, info, decryption_info
@@ -436,6 +434,15 @@ class Portal(DBPortal, BasePortal):
                 self.avatar_set = False
         return True
 
+    async def update_info_from_puppet(self, puppet: p.Puppet | None = None) -> bool:
+        if not self.is_direct:
+            return False
+        if not puppet:
+            puppet = await self.get_dm_puppet()
+        changed = await self._update_name(puppet.name)
+        changed = await self._update_photo_from_puppet(puppet) or changed
+        return changed
+
     """
     async def sync_per_room_nick(self, puppet: p.Puppet, name: str) -> None:
         intent = puppet.intent_for(self)
@@ -457,31 +464,33 @@ class Portal(DBPortal, BasePortal):
             )
     """
 
+    async def _update_participant(
+        self, source: u.User, participant: UserInfoUnion
+    ) -> bool:
+        # TODO nick map?
+        self.log.trace(f"Syncing participant {participant.userId}")
+        puppet = await p.Puppet.get_by_ktid(participant.userId)
+        await puppet.update_info_from_participant(source, participant)
+        changed = False
+        if self.is_direct and self._kt_sender == puppet.ktid and self.encrypted:
+            changed = await self._update_info_from_puppet(puppet.name) or changed
+        if self.mxid:
+            if puppet.ktid != self.kt_receiver or puppet.is_real_user:
+                await puppet.intent_for(self).ensure_joined(self.mxid, bot=self.main_intent)
+            #if puppet.ktid in nick_map:
+            #    await self.sync_per_room_nick(puppet, nick_map[puppet.ktid])
+        return changed
+
+
     async def _update_participants(self, source: u.User, participants: list[UserInfoUnion] | None = None) -> bool:
+        # TODO nick map?
         if participants is None:
             self.log.debug("Called _update_participants with no participants, fetching them now...")
             participants = await source.client.get_participants(self.channel_props)
-        changed = False
-        if not self._main_intent:
-            assert self.is_direct, "_main_intent for non-direct chat portal should have been set already"
-            self._kt_sender = participants[
-                0 if self.kt_type == KnownChannelType.MemoChat or participants[0].userId != source.ktid else 1
-            ].userId
-            self._main_intent = (await p.Puppet.get_by_ktid(self._kt_sender)).default_mxid_intent
-        else:
-            self._kt_sender = (await p.Puppet.get_by_mxid(self._main_intent.mxid)).ktid if self.is_direct else None
-        # TODO nick_map?
-        for participant in participants:
-            puppet = await p.Puppet.get_by_ktid(participant.userId)
-            await puppet.update_info_from_participant(source, participant)
-            if self.is_direct and self._kt_sender == puppet.ktid and self.encrypted:
-                changed = await self._update_name(puppet.name) or changed
-                changed = await self._update_photo_from_puppet(puppet) or changed
-            if self.mxid:
-                if puppet.ktid != self.kt_receiver or puppet.is_real_user:
-                    await puppet.intent_for(self).ensure_joined(self.mxid, bot=self.main_intent)
-                #if puppet.ktid in nick_map:
-                #    await self.sync_per_room_nick(puppet, nick_map[puppet.ktid])
+        sync_tasks = [
+            self._update_participant(source, pcp) for pcp in participants
+        ]
+        changed = any(await asyncio.gather(*sync_tasks))
         return changed
 
     # endregion
@@ -833,7 +842,7 @@ class Portal(DBPortal, BasePortal):
         if message.relates_to.rel_type == RelationType.REPLY:
             reply_to_msg = await DBMessage.get_by_mxid(message.relates_to.event_id, self.mxid)
             if reply_to_msg:
-                reply_to = reply_to_msg.fbid
+                reply_to = reply_to_msg.ktid
             else:
                 self.log.warning(
                     f"Couldn't find reply target {message.relates_to.event_id}"
@@ -928,16 +937,6 @@ class Portal(DBPortal, BasePortal):
         )
         self._typing = users
 
-    async def enable_dm_encryption(self) -> bool:
-        ok = await super().enable_dm_encryption()
-        if ok:
-            try:
-                puppet = await p.Puppet.get_by_ktid(self.ktid)
-                await self.main_intent.set_room_name(self.mxid, puppet.name)
-            except Exception:
-                self.log.warning(f"Failed to set room name", exc_info=True)
-        return ok
-
     # endregion
     # region KakaoTalk event handling
 
@@ -956,6 +955,38 @@ class Portal(DBPortal, BasePortal):
                 return False
         return True
 
+    async def _add_remote_reply(
+        self, content: MessageEventContent, reply_to: ReplyAttachment
+    ) -> None:
+        message = await DBMessage.get_by_ktid(reply_to.src_logId, self.kt_receiver)
+        if not message:
+            self.log.warning(
+                f"Couldn't find reply target {reply_to.src_logId} to bridge reply metadata to Matrix"
+            )
+            return
+
+        content.set_reply(message.mxid)
+        if not isinstance(content, TextMessageEventContent):
+            return
+
+        try:
+            evt = await self.main_intent.get_event(message.mx_room, message.mxid)
+        except (MNotFound, MForbidden):
+            evt = None
+        if not evt:
+            return
+
+        if evt.type == EventType.ROOM_ENCRYPTED:
+            try:
+                evt = await self.matrix.e2ee.decrypt(evt, wait_session_timeout=0)
+            except SessionNotFound:
+                return
+
+        if isinstance(evt.content, TextMessageEventContent):
+            evt.content.trim_reply_fallback()
+
+        content.set_reply(evt)
+
     async def handle_remote_message(
         self,
         source: u.User,
@@ -976,6 +1007,7 @@ class Portal(DBPortal, BasePortal):
         sender: p.Puppet,
         message: Chatlog,
     ) -> None:
+        # TODO Backfill!! This avoids timing conflicts on startup sync
         self.log.debug(f"Handling KakaoTalk event {message.logId}")
         if not self.mxid:
             mxid = await self.create_matrix_room(source)
@@ -996,25 +1028,25 @@ class Portal(DBPortal, BasePortal):
             await intent.ensure_joined(self.mxid)
             self._backfill_leave.add(intent)
 
-        handler = self._message_type_handler_map.get(message.type)
-        if not handler:
-            self.log.warning(f"No handler for message type {message.type}, falling back to text")
-            handler = Portal._handle_remote_text
+        handler = self._message_type_handler_map.get(message.type, Portal._handle_remote_unsupported)
         event_ids = [
             event_id for event_id in 
             await handler(
                 self,
-                source,
-                intent,
-                message.attachment,
-                message.sendAt,
-                message.text)
+                source=source,
+                intent=intent,
+                attachment=message.attachment,
+                timestamp=message.sendAt,
+                message_text=message.text,
+                message_type=message.type,
+            )
             if event_id
         ]
         if not event_ids:
             self.log.warning(f"Unhandled KakaoTalk message {message.logId}")
             return
         self.log.debug(f"Handled KakaoTalk message {message.logId} -> {event_ids}")
+        # TODO Might have to handle remote reactions on messages created by bulk_create
         await DBMessage.bulk_create(
             ktid=message.logId,
             kt_chat=self.ktid,
@@ -1025,44 +1057,74 @@ class Portal(DBPortal, BasePortal):
         )
         await self._send_delivery_receipt(event_ids[-1])
 
-    async def _handle_remote_text(
+    async def _handle_remote_unsupported(
         self,
-        source: u.User,
         intent: IntentAPI,
-        attachment: None,
         timestamp: int,
         message_text: str | None,
+        message_type: ChatType,
+        **_
+    ) -> Awaitable[list[EventID]]:
+        try:
+            type_str = KnownChatType(message_type).name.lower()
+        except ValueError:
+            type_str = str(message_type)
+        self.log.warning("No handler for message type \"%s\" (%s)",
+            type_str,
+            f"text = {message_text}" if message_text is not None else "no text",
+        )
+        if message_text:
+            events = await self._handle_remote_text(
+                intent=intent,
+                timestamp=timestamp,
+                message_text=message_text,
+            )
+        else:
+            events = []
+        content = TextMessageEventContent(
+            msgtype=MessageType.NOTICE,
+            body=f"\u26a0 Unbridgeable message ({type_str})",
+        )
+        if events:
+            content.set_reply(events[-1])
+        events.append(await self._send_message(intent, content, timestamp=timestamp))
+        return events
+
+    async def _handle_remote_text(
+        self,
+        intent: IntentAPI,
+        timestamp: int,
+        message_text: str | None,
+        **_
     ) -> list[EventID]:
         # TODO Handle mentions properly
         content = await kakaotalk_to_matrix(message_text)
-        # TODO Replies
         return [await self._send_message(intent, content, timestamp=timestamp)]
 
-    def _handle_remote_photo(
+    async def _handle_remote_reply(
         self,
-        source: u.User,
         intent: IntentAPI,
-        attachment: PhotoAttachment,
+        attachment: ReplyAttachment,
         timestamp: int,
-        message_text: str | None,
-    ) -> Awaitable[list[EventID]]:
-        return asyncio.gather(self._handle_remote_uniphoto(
-            source, intent, attachment, timestamp, message_text
-        ))
+        message_text: str,
+        **_
+    ) -> list[EventID]:
+        content = await kakaotalk_to_matrix(message_text)
+        await self._add_remote_reply(content, attachment)
+        return [await self._send_message(intent, content, timestamp=timestamp)]
+
+    def _handle_remote_photo(self, **kwargs) -> Awaitable[list[EventID]]:
+        return asyncio.gather(self._handle_remote_uniphoto(**kwargs))
 
     async def _handle_remote_multiphoto(
         self,
-        source: u.User,
-        intent: IntentAPI,
         attachment: MultiPhotoAttachment,
-        timestamp: int,
-        message_text: str | None,
+        **kwargs
     ) -> Awaitable[list[EventID]]:
         # TODO Upload media concurrently, but post messages sequentially
         return [
             await self._handle_remote_uniphoto(
-                source, intent,
-                PhotoAttachment(
+                attachment=PhotoAttachment(
                     shout=attachment.shout,
                     mentions=attachment.mentions,
                     urls=attachment.urls,
@@ -1077,21 +1139,18 @@ class Portal(DBPortal, BasePortal):
                     cs=attachment.csl[i],
                     mt=attachment.mtl[i],
                 ),
-                timestamp, message_text,
+                **kwargs
             )
             for i in range(len(attachment.imageUrls))
         ]
 
     def _handle_remote_uniphoto(
         self,
-        source: u.User,
-        intent: IntentAPI,
         attachment: PhotoAttachment,
-        timestamp: int,
-        message_text: str | None,
+        **kwargs
     ) -> Awaitable[EventID]:
         return self._handle_remote_media(
-            source, intent, attachment, timestamp, message_text,
+            attachment,
             ImageInfo(
                 mimetype=attachment.mt,
                 size=attachment.s,
@@ -1099,70 +1158,67 @@ class Portal(DBPortal, BasePortal):
                 height=attachment.h,
             ),
             MessageType.IMAGE,
+            **kwargs
         )
 
     def _handle_remote_video(
         self,
-        source: u.User,
-        intent: IntentAPI,
         attachment: VideoAttachment,
-        timestamp: int,
-        message_text: str | None,
+        **kwargs
     ) -> Awaitable[list[EventID]]:
         return asyncio.gather(self._handle_remote_media(
-            source, intent, attachment, timestamp, message_text,
+            attachment,
             VideoInfo(
                 duration=attachment.d,
                 width=attachment.w,
                 height=attachment.h,
             ),
             MessageType.VIDEO,
+            **kwargs
         ))
 
     def _handle_remote_audio(
         self,
-        source: u.User,
-        intent: IntentAPI,
         attachment: AudioAttachment,
-        timestamp: int,
-        message_text: str | None,
+        **kwargs
     ) -> Awaitable[list[EventID]]:
         return asyncio.gather(self._handle_remote_media(
-            source, intent, attachment, timestamp, message_text,
+            attachment,
             AudioInfo(
                 size=attachment.s,
                 duration=attachment.d,
             ),
             MessageType.AUDIO,
+            **kwargs
         ))
 
     """ TODO Find what auth is required for reading file contents
     def _handle_remote_file(
         self,
-        source: u.User,
-        intent: IntentAPI,
         attachment: FileAttachment,
-        timestamp: int,
-        message_text: str | None,
+        **kwargs
     ) -> Awaitable[list[EventID]]:
         return asyncio.gather(self._handle_remote_media(
-            source, intent, attachment, timestamp, message_text,
+            attachment,
             FileInfo(
                 size=attachment.size,
             ),
             MessageType.FILE,
+            **kwargs
         ))
     """
 
     async def _handle_remote_media(
         self,
-        source: u.User,
-        intent: IntentAPI,
         attachment: MediaAttachment,
-        timestamp: int,
-        message_text: str | None,
         info: MediaInfo,
         msgtype: MessageType,
+        *,
+        source: u.User,
+        intent: IntentAPI,
+        timestamp: int,
+        message_text: str | None,
+        **_
     ) -> EventID:
         mxc, additional_info, decryption_info = await self._reupload_remote_file(
             attachment.url,
@@ -1177,7 +1233,6 @@ class Portal(DBPortal, BasePortal):
         content = MediaMessageEventContent(
             url=mxc, file=decryption_info, msgtype=msgtype, body=message_text, info=info
         )
-        # TODO Replies
         return await self._send_message(intent, content, timestamp=timestamp)
 
     # TODO Many more remote handlers
@@ -1232,7 +1287,7 @@ class Portal(DBPortal, BasePortal):
         messages = await source.client.get_chats(
             self.channel_props,
             after_log_id,
-            limit
+            limit,
         )
         if not messages:
             self.log.debug("Didn't get any messages from server")
@@ -1251,21 +1306,25 @@ class Portal(DBPortal, BasePortal):
     # region Database getters
 
     async def postinit(self) -> None:
-        self.by_ktid[self._ktid_full] = self
+        self.by_ktid[self.ktid_full] = self
         if self.mxid:
             self.by_mxid[self.mxid] = self
         if not self.is_direct:
             self._main_intent = self.az.intent
-        elif self.mxid:
+        else:
             # 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)
+            if self.kt_type == KnownChannelType.MemoChat:
+                self._kt_sender = user.ktid
             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")
+                # NOTE This throws if the user isn't connected--good!
+                #      Nothing should init a portal for a disconnected user.
+                participants = await user.client.get_participants(self.channel_props)
+                self._kt_sender = participants[
+                    0 if participants[0].userId != user.ktid else 1
+                ].userId
+            self._main_intent = (await p.Puppet.get_by_ktid(self._kt_sender)).default_mxid_intent
 
     @classmethod
     @async_getter_lock

matrix_appservice_kakaotalk/puppet.py

@@ -204,8 +204,9 @@ class Puppet(DBPuppet, BasePuppet):
             return True
         return False
 
-    @staticmethod
+    @classmethod
     async def reupload_avatar(
+        cls,
         source: u.User,
         intent: IntentAPI,
         url: str,
@@ -214,7 +215,9 @@ class Puppet(DBPuppet, BasePuppet):
         async with source.client.get(url) as resp:
             data = await resp.read()
         mime = magic.mimetype(data)
-        return await intent.upload_media(data, mime_type=mime)
+        return await intent.upload_media(
+            data, mime_type=mime, async_upload=cls.config["homeserver.async_media"]
+        )
 
     async def _update_photo(self, source: u.User, photo_id: str) -> bool:
         if photo_id != self.photo_id or not self.avatar_set:

matrix_appservice_kakaotalk/user.py

@@ -40,7 +40,7 @@ from .kt.client.errors import AuthenticationRequired, ResponseError
 from .kt.types.api.struct.profile import ProfileStruct
 from .kt.types.bson import Long
 from .kt.types.channel.channel_info import ChannelInfo, NormalChannelInfo, NormalChannelData
-from .kt.types.channel.channel_type import ChannelType
+from .kt.types.channel.channel_type import ChannelType, KnownChannelType
 from .kt.types.chat.chat import Chatlog
 from .kt.types.client.client_session import LoginDataItem, LoginResult
 from .kt.types.oauth import OAuthCredential
@@ -555,6 +555,30 @@ class User(DBUser, BaseUser):
             return None
         return await pu.Puppet.get_by_ktid(self.ktid)
 
+    async def get_portal_with(self, puppet: pu.Puppet, create: bool = True) -> po.Portal | None:
+        # TODO Make upstream request to return custom failure message
+        if not self.ktid or not self.is_connected:
+            return None
+        if puppet.ktid != self.ktid:
+            kt_type = KnownChannelType.DirectChat
+            ktid = await self.client.get_friend_dm_id(puppet.ktid)
+        else:
+            kt_type = KnownChannelType.MemoChat
+            memo_ids = await self.client.get_memo_ids()
+            if not memo_ids:
+                ktid = Long(0)
+            else:
+                ktid = memo_ids[0]
+                if len(memo_ids) > 1:
+                    self.log.info("Found multiple memo chats, so using the first one as a fallback")
+        if ktid:
+            return await po.Portal.get_by_ktid(
+                ktid, kt_receiver=self.ktid, create=create, kt_type=kt_type
+            )
+        else:
+            self.log.warning(f"Didn't find an existing DM channel with KakaoTalk user {puppet.ktid}, so not creating one")
+            return None
+
     # region KakaoTalk event handling
 
     async def on_connect(self) -> None:

node/src/client.js

@@ -73,6 +73,38 @@ class UserClient {
 
 		this.mxid = mxid
 		this.peerClient = peerClient
+
+		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,
+			})
+		})
 	}
 
 	/**
@@ -127,42 +159,7 @@ class UserClient {
 	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
+		return await this.#talkClient.login(credential)
 	}
 
 	disconnect() {
@@ -346,6 +343,14 @@ export default class PeerClient {
 		return this.userClients.get(mxid)
 	}
 
+	/**
+	 * @param {string} mxid
+	 * @param {Object} channel_props
+	 */
+	async #getUserChannel(mxid, channel_props) {
+		return await this.#getUser(mxid).getChannel(channel_props)
+	}
+
 	/**
 	 * @param {Object} req
 	 * @param {OAuthCredential} req.oauth_credential
@@ -420,8 +425,7 @@ export default class PeerClient {
 	 * @param {Object} req.channel_props
 	 */
 	getPortalChannelInfo = async (req) => {
-		const userClient = this.#getUser(req.mxid)
-		const talkChannel = await userClient.getChannel(req.channel_props)
+		const talkChannel = await this.#getUserChannel(req.mxid, req.channel_props)
 
 		const res = await talkChannel.updateAll()
 		if (!res.success) return res
@@ -439,8 +443,7 @@ export default class PeerClient {
 	 * @param {Object} req.channel_props
 	 */
 	getParticipants = async (req) => {
-		const userClient = this.#getUser(req.mxid)
-		const talkChannel = await userClient.getChannel(req.channel_props)
+		const talkChannel = await this.#getUserChannel(req.mxid, req.channel_props)
 		return await talkChannel.getAllLatestUserInfo()
 	}
 
@@ -452,8 +455,7 @@ export default class PeerClient {
 	 * @param {?Number} req.limit
 	 */
 	getChats = async (req) => {
-		const userClient = this.#getUser(req.mxid)
-		const talkChannel = await userClient.getChannel(req.channel_props)
+		const talkChannel = await this.#getUserChannel(req.mxid, req.channel_props)
 
 		const res = await talkChannel.getChatListFrom(req.sync_from)
 		if (res.success && 0 < req.limit && req.limit < res.result.length) {
@@ -464,22 +466,49 @@ export default class PeerClient {
 
 	/**
 	 * @param {Object} req
-	 * @param {?string} req.mxid
-	 * @param {?OAuthCredential} req.oauth_credential
+	 * @param {string} req.mxid
 	 */
 	listFriends = async (req) => {
 		return await this.#getUser(req.mxid).serviceClient.requestFriendList()
 	}
 
+	/**
+	 * @param {Object} req
+	 * @param {string} req.mxid The user whose friend is being looked up.
+	 * @param {string} req.friend_id The friend to search for.
+	 * @param {string} propertyName The property to retrieve from the specified friend.
+	 */
+	getFriendProperty = async (req, propertyName) => {
+		const res = await this.#getUser(req.mxid).serviceClient.findFriendById(req.friend_id)
+		if (!res.success) return res
+
+		return this.#makeCommandResult(res.result.friend[propertyName])
+	}
+
+	/**
+	 * @param {Object} req
+	 * @param {string} req.mxid
+	 */
+	getMemoIds = (req) => {
+		/** @type Long[] */
+		const channelIds = []
+		const channelList = this.#getUser(req.mxid).talkClient.channelList
+		for (const channel of channelList.all()) {
+			if (channel.info.type == "MemoChat") {
+				channelIds.push(channel.channelId)
+			}
+		}
+		return channelIds
+	}
+
 	/**
 	 * @param {Object} req
 	 * @param {string} req.mxid
 	 * @param {Object} req.channel_props
 	 * @param {string} req.text
 	 */
-	sendMessage = async (req) => {
-		const userClient = this.#getUser(req.mxid)
-		const talkChannel = await userClient.getChannel(req.channel_props)
+	sendChat = async (req) => {
+		const talkChannel = await this.#getUserChannel(req.mxid, req.channel_props)
 
 		return await talkChannel.sendChat({
 			type: KnownChatType.TEXT,
@@ -499,8 +528,7 @@ export default class PeerClient {
 	 * @param {?string} req.ext
 	 */
 	sendMedia = async (req) => {
-		const userClient = this.#getUser(req.mxid)
-		const talkChannel = await userClient.getChannel(req.channel_props)
+		const talkChannel = await this.#getUserChannel(req.mxid, req.channel_props)
 
 		return await talkChannel.sendMedia(req.type, {
 			data: Uint8Array.from(req.data),
@@ -590,7 +618,9 @@ export default class PeerClient {
 				get_participants: this.getParticipants,
 				get_chats: this.getChats,
 				list_friends: this.listFriends,
-				send_message: this.sendMessage,
+				get_friend_dm_id: req => this.getFriendProperty(req, "directChatId"),
+				get_memo_ids: this.getMemoIds,
+				send_chat: this.sendChat,
 				send_media: this.sendMedia,
 			}[req.command] || this.handleUnknownCommand
 		}

requirements.txt

@@ -1,7 +1,7 @@
 aiohttp>=3,<4
 asyncpg>=0.20,<0.26
 commonmark>=0.8,<0.10
-mautrix==0.15.0rc4
+mautrix>=0.15.4,<0.16
 pycryptodome>=3,<4
 python-magic>=0.4,<0.5
 ruamel.yaml>=0.15.94,<0.18

setup.py

@@ -48,7 +48,7 @@ setuptools.setup(
 
     install_requires=install_requires,
     extras_require=extras_require,
-    python_requires="~=3.7",
+    python_requires="~=3.8",
 
     classifiers=[
         "Development Status :: 1 - Planning",