Implement message redaction/unsend in both directions

ROADMAP.md

@@ -9,7 +9,7 @@
       * [ ] Voice messages
       * [ ] Locations
       * [ ] †Files
-  * [ ] Message redactions
+  * [x] Message redactions
   * [x] Message reactions
   * [ ] Presence
   * [ ] Typing notifications
@@ -22,7 +22,7 @@
       * [x] Videos
       * [ ] Voice messages
       * [ ] Locations
-  * [ ] Message unsend
+  * [x] Message unsend
   * [ ] Message reactions
   * [ ] Message history
   * [ ] Presence

mauigpapi/http/thread.py

@@ -70,3 +70,7 @@ class ThreadAPI(BaseAndroidAPI):
             for item in resp.thread.items:
                 yield item
 
+    async def delete_item(self, thread_id: str, item_id: str) -> None:
+        await self.std_http_post(f"/api/v1/direct_v2/threads/{thread_id}/items/{item_id}/delete/",
+                                 data={"_csrftoken": self.state.cookies.csrf_token,
+                                       "_uuid": self.state.device.uuid})

mauigpapi/mqtt/conn.py

@@ -223,9 +223,9 @@ class AndroidMQTT:
             subitem_key = rest[0]
             if subitem_key == "approval_required_for_new_members":
                 additional["approval_required_for_new_members"] = True
-            elif subitem_key == ["participants"]:
+            elif subitem_key == "participants":
                 additional["participants"] = {rest[1]: rest[2]}
-            elif subitem_key == ["items"]:
+            elif subitem_key == "items":
                 additional["item_id"] = rest[1]
                 # TODO wtf is this?
                 #      it has something to do with reactions
@@ -235,6 +235,7 @@ class AndroidMQTT:
                     }
             elif subitem_key in ("admin_user_ids", "activity_indicator_id"):
                 additional[subitem_key] = rest[1]
+        print("Parsed path", path, "->", additional)
         return additional
 
     def _on_message_sync(self, payload: bytes) -> None:
@@ -253,7 +254,7 @@ class AndroidMQTT:
                         **raw_message,
                         **json.loads(part.value),
                     }
-                except json.JSONDecodeError:
+                except (json.JSONDecodeError, TypeError):
                     raw_message["value"] = part.value
                 message = MessageSyncMessage.deserialize(raw_message)
                 evt = MessageSyncEvent(iris=parsed_item, message=message)

mauigpapi/types/thread_item.py

@@ -276,7 +276,7 @@ class AnimatedMediaItem(SerializableAttrs['AnimatedMediaItem']):
 class ThreadItem(SerializableAttrs['ThreadItem']):
     item_id: Optional[str] = None
     user_id: Optional[int] = None
-    timestamp: int
+    timestamp: Optional[int] = None
     item_type: Optional[ThreadItemType] = None
     is_shh_mode: bool = False
 

mautrix_instagram/db/message.py

@@ -31,10 +31,12 @@ class Message:
     mx_room: RoomID
     item_id: str
     receiver: int
+    sender: int
 
     async def insert(self) -> None:
-        q = "INSERT INTO message (mxid, mx_room, item_id, receiver) VALUES ($1, $2, $3, $4)"
-        await self.db.execute(q, self.mxid, self.mx_room, self.item_id, self.receiver)
+        q = ("INSERT INTO message (mxid, mx_room, item_id, receiver, sender) "
+             "VALUES ($1, $2, $3, $4, $5)")
+        await self.db.execute(q, self.mxid, self.mx_room, self.item_id, self.receiver, self.sender)
 
     async def delete(self) -> None:
         q = "DELETE FROM message WHERE item_id=$1 AND receiver=$2"
@@ -46,7 +48,7 @@ class Message:
 
     @classmethod
     async def get_by_mxid(cls, mxid: EventID, mx_room: RoomID) -> Optional['Message']:
-        row = await cls.db.fetchrow("SELECT mxid, mx_room, item_id, receiver "
+        row = await cls.db.fetchrow("SELECT mxid, mx_room, item_id, receiver, sender "
                                     "FROM message WHERE mxid=$1 AND mx_room=$2", mxid, mx_room)
         if not row:
             return None
@@ -54,7 +56,7 @@ class Message:
 
     @classmethod
     async def get_by_item_id(cls, item_id: str, receiver: int = 0) -> Optional['Message']:
-        row = await cls.db.fetchrow("SELECT mxid, mx_room, item_id, receiver "
+        row = await cls.db.fetchrow("SELECT mxid, mx_room, item_id, receiver, sender "
                                     "FROM message WHERE item_id=$1 AND receiver=$2",
                                     item_id, receiver)
         if not row:

mautrix_instagram/db/upgrade.py

@@ -64,6 +64,7 @@ async def upgrade_v1(conn: Connection) -> None:
         mx_room  TEXT NOT NULL,
         item_id  TEXT,
         receiver BIGINT,
+        sender   BIGINT NOT NULL,
         PRIMARY KEY (item_id, receiver),
         UNIQUE (mxid, mx_room)
     )""")

mautrix_instagram/portal.py

@@ -23,13 +23,14 @@ import asyncio
 import magic
 from yarl import URL
 
-from mauigpapi.types import Thread, ThreadUser, ThreadItem, RegularMediaItem, MediaType
+from mauigpapi.types import (Thread, ThreadUser, ThreadItem, RegularMediaItem, MediaType,
+                             ReactionStatus)
 from mautrix.appservice import AppService, IntentAPI
 from mautrix.bridge import BasePortal, NotificationDisabler
 from mautrix.types import (EventID, MessageEventContent, RoomID, EventType, MessageType, ImageInfo,
                            VideoInfo, MediaMessageEventContent, TextMessageEventContent,
                            ContentURI, EncryptedFile)
-from mautrix.errors import MatrixError
+from mautrix.errors import MatrixError, MForbidden
 from mautrix.util.simple_lock import SimpleLock
 from mautrix.util.network_retry import call_with_net_retry
 
@@ -161,6 +162,7 @@ class Portal(DBPortal, BasePortal):
             mime_type = message.info.mimetype or magic.from_buffer(data, mime=True)
             if mime_type == "image/jpeg":
                 upload_resp = await sender.client.upload_jpeg_photo(data)
+                # TODO I don't think this works
                 resp = await sender.mqtt.send_media(self.thread_id, upload_resp.upload_id,
                                                     client_context=request_id)
             else:
@@ -173,7 +175,7 @@ class Portal(DBPortal, BasePortal):
         else:
             self._msgid_dedup.appendleft(resp.payload.item_id)
             await DBMessage(mxid=event_id, mx_room=self.mxid, item_id=resp.payload.item_id,
-                            receiver=self.receiver).insert()
+                            receiver=self.receiver, sender=sender.igpk).insert()
             self._reqid_dedup.remove(request_id)
             await self._send_delivery_receipt(event_id)
             self.log.debug(f"Handled Matrix message {event_id} -> {resp.payload.item_id}")
@@ -205,16 +207,26 @@ class Portal(DBPortal, BasePortal):
             return
 
         # TODO implement
-        # reaction = await DBReaction.get_by_mxid(event_id, self.mxid)
-        # if reaction:
-        #     try:
-        #         await reaction.delete()
-        #         await sender.client.conversation(self.twid).delete_reaction(reaction.tw_msgid,
-        #                                                                     reaction.reaction)
-        #         await self._send_delivery_receipt(redaction_event_id)
-        #         self.log.trace(f"Removed {reaction} after Matrix redaction")
-        #     except Exception:
-        #         self.log.exception("Removing reaction failed")
+        reaction = await DBReaction.get_by_mxid(event_id, self.mxid)
+        if reaction:
+            try:
+                await reaction.delete()
+                await sender.mqtt.send_reaction(self.thread_id, item_id=reaction.ig_item_id,
+                                                reaction_status=ReactionStatus.DELETED, emoji="")
+                await self._send_delivery_receipt(redaction_event_id)
+                self.log.trace(f"Removed {reaction} after Matrix redaction")
+            except Exception:
+                self.log.exception("Removing reaction failed")
+            return
+
+        message = await DBMessage.get_by_mxid(event_id, self.mxid)
+        if message:
+            try:
+                await message.delete()
+                await sender.client.delete_item(self.thread_id, message.item_id)
+                self.log.trace(f"Removed {message} after Matrix redaction")
+            except Exception:
+                self.log.exception("Removing message failed")
 
     async def handle_matrix_leave(self, user: 'u.User') -> None:
         if self.is_direct:
@@ -305,12 +317,24 @@ class Portal(DBPortal, BasePortal):
             # TODO handle attachments and reactions
             if event_id:
                 await DBMessage(mxid=event_id, mx_room=self.mxid, item_id=item.item_id,
-                                receiver=self.receiver).insert()
+                                receiver=self.receiver, sender=sender.pk).insert()
                 await self._send_delivery_receipt(event_id)
                 self.log.debug(f"Handled Instagram message {item.item_id} -> {event_id}")
             else:
                 self.log.debug(f"Unhandled Instagram message {item.item_id}")
 
+    async def handle_instagram_remove(self, item_id: str) -> None:
+        message = await DBMessage.get_by_item_id(item_id, self.receiver)
+        if message is None:
+            return
+        await message.delete()
+        sender = await p.Puppet.get_by_pk(message.sender)
+        try:
+            await sender.intent_for(self).redact(self.mxid, message.mxid)
+        except MForbidden:
+            await self.main_intent.redact(self.mxid, message.mxid)
+        self.log.debug(f"Redacted {message.mxid} after Instagram unsend")
+
     # endregion
     # region Updating portal info
 

mautrix_instagram/user.py

@@ -19,11 +19,9 @@ from collections import defaultdict
 import asyncio
 import logging
 
-from mauigpapi.mqtt import (AndroidMQTT, Connect, Disconnect, GraphQLSubscription,
-                            SkywalkerSubscription)
-from mauigpapi.http import AndroidAPI
-from mauigpapi.state import AndroidState
-from mauigpapi.types import CurrentUser, MessageSyncEvent
+from mauigpapi import AndroidAPI, AndroidState, AndroidMQTT
+from mauigpapi.mqtt import Connect, Disconnect, GraphQLSubscription, SkywalkerSubscription
+from mauigpapi.types import CurrentUser, MessageSyncEvent, Operation
 from mauigpapi.errors import IGNotLoggedInError
 from mautrix.bridge import BaseUser
 from mautrix.types import UserID, RoomID, EventID, TextMessageEventContent, MessageType
@@ -236,16 +234,21 @@ class User(DBUser, BaseUser):
 
     @async_time(METRIC_MESSAGE)
     async def handle_message(self, evt: MessageSyncEvent) -> None:
-        # We don't care about messages with no sender
-        if not evt.message.user_id:
-            return
         portal = await po.Portal.get_by_thread_id(evt.message.thread_id, receiver=self.igpk)
         if not portal.mxid:
             # TODO try to find the thread?
             self.log.warning(f"Ignoring message to unknown thread {evt.message.thread_id}")
             return
-        sender = await pu.Puppet.get_by_pk(evt.message.user_id)
-        await portal.handle_instagram_item(self, sender, evt.message)
+        self.log.trace(f"Received message sync event {evt.message}")
+        sender = await pu.Puppet.get_by_pk(evt.message.user_id) if evt.message.user_id else None
+        if evt.message.op == Operation.ADD:
+            if not sender:
+                # I don't think we care about adds with no sender
+                return
+            await portal.handle_instagram_item(self, sender, evt.message)
+        elif evt.message.op == Operation.REMOVE:
+            # Removes don't have a sender, only the message sender can unsend messages anyway
+            await portal.handle_instagram_remove(evt.message.item_id)
 
     # @async_time(METRIC_RECEIPT)
     # async def handle_receipt(self, evt: ConversationReadEntry) -> None: