39 add support to reactions and stickers

gupshup_matrix/db/__init__.py

@@ -4,12 +4,13 @@
 from .message import Message
 from .portal import Portal
 from .puppet import Puppet
+from .reaction import Reaction
 from .upgrade import upgrade_table
 from .user import User
 
 
 def init(db: Database) -> None:
-    for table in (Puppet, Portal, User, Message, GupshupApplication):
+    for table in (Puppet, Portal, User, Message, GupshupApplication, Reaction):
         table.db = db
 
 

gupshup_matrix/db/reaction.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, ClassVar, Optional
+
+import asyncpg
+from attr import dataclass
+from mautrix.types import EventID, RoomID, UserID
+from mautrix.util.async_db import Database
+
+fake_db = Database.create("") if TYPE_CHECKING else None
+
+log: logging.Logger = logging.getLogger("meta.out")
+
+
+@dataclass
+class Reaction:
+    db: ClassVar[Database] = fake_db
+
+    event_mxid: EventID
+    room_id: RoomID
+    sender: UserID
+    gs_message_id: str
+    reaction: str
+    created_at: float
+
+    @property
+    def _values(self):
+        return (
+            self.event_mxid,
+            self.room_id,
+            self.sender,
+            self.gs_message_id,
+            self.reaction,
+            self.created_at,
+        )
+
+    _columns = "event_mxid, room_id, sender, gs_message_id, reaction, created_at"
+
+    async def insert(self) -> None:
+        q = f"INSERT INTO reaction ({self._columns}) VALUES ($1, $2, $3, $4, $5, $6)"
+        await self.db.execute(q, *self._values)
+
+    @classmethod
+    def _from_row(cls, row: asyncpg.Record) -> Optional["Reaction"]:
+        return cls(**row)
+
+    @classmethod
+    async def delete_all(cls, room_id: RoomID) -> None:
+        await cls.db.execute("DELETE FROM reaction WHERE room_id=$1", room_id)
+
+    @classmethod
+    async def get_by_gs_message_id(
+        cls, gs_message_id: str, sender: UserID
+    ) -> Optional["Reaction"]:
+        q = f"SELECT {cls._columns} FROM reaction WHERE gs_message_id=$1 AND sender=$2"
+        row = await cls.db.fetchrow(q, gs_message_id, sender)
+        if not row:
+            return None
+        return cls._from_row(row)
+
+    @classmethod
+    async def get_by_event_mxid(cls, event_mxid: EventID, room_id: RoomID) -> Optional["Reaction"]:
+        q = f"SELECT {cls._columns} FROM reaction WHERE event_mxid=$1 AND room_id=$2"
+        row = await cls.db.fetchrow(q, event_mxid, room_id)
+        if not row:
+            return None
+        return cls._from_row(row)
+
+    @classmethod
+    async def get_last_reaction(cls, room_id: RoomID) -> "Reaction":
+        q = f"""
+            SELECT {cls._columns}
+            FROM reaction WHERE room_id=$1 ORDER BY created_at DESC LIMIT 1
+        """
+        row = await cls.db.fetchrow(q, room_id)
+        if not row:
+            return None
+        return cls._from_row(row)
+
+    @classmethod
+    async def delete_by_event_mxid(
+        cls, event_mxid: EventID, room_id: RoomID, sender: UserID
+    ) -> "Reaction":
+        q = "DELETE FROM reaction WHERE event_mxid=$1 AND room_id=$2 AND sender=$3"
+        row = await cls.db.fetchrow(q, event_mxid, room_id, sender)
+        if not row:
+            return None
+        return cls._from_row(row)

gupshup_matrix/db/upgrade.py

@@ -62,3 +62,23 @@ async def upgrade_v1(conn: Connection) -> None:
 @upgrade_table.register(description="Add field encrypted to portal table")
 async def upgrade_v2(conn: Connection) -> None:
     await conn.execute("ALTER TABLE portal ADD COLUMN encrypted BOOLEAN DEFAULT false")
+
+
+@upgrade_table.register(description="Add reaction table to store reactions")
+async def upgrade_v3(conn: Connection) -> None:
+    await conn.execute(
+        """CREATE TABLE reaction (
+        event_mxid      VARCHAR(255) PRIMARY KEY,
+        room_id         VARCHAR(255) NOT NULL,
+        sender          VARCHAR(255) NOT NULL,
+        gs_message_id   TEXT NOT NULL,
+        reaction        VARCHAR(255),
+        created_at      TIMESTAMP WITH TIME ZONE NOT NULL
+    )"""
+    )
+    await conn.execute("ALTER TABLE message ADD CONSTRAINT unique_gsid UNIQUE (gsid);")
+    await conn.execute(
+        """ALTER TABLE reaction ADD CONSTRAINT FK_message_gsid
+        FOREIGN KEY (gs_message_id) references message (gsid)
+        ON DELETE CASCADE"""
+    )

gupshup_matrix/example-config.yaml

@@ -124,6 +124,8 @@ gupshup:
     # Gupsghup base URL
     base_url: https://api.gupshup.io/sm/api/v1/msg
     read_url: https://api.gupshup.io/wa/app/{appId}/msg/{msgId}/read
+    # Gupshup reaction URL
+    cloud_api_url: https://api.gupshup.io/wa/api/v1/msg
     # Path prefix for webhook endpoints. Subpaths are /status and /receive.
     # Note that the webhook must be put behind a reverse proxy with https.
     webhook_path: /gupshup

gupshup_matrix/gupshup/api.py

@@ -18,6 +18,7 @@ class GupshupClient:
     def __init__(self, config: Config, loop: asyncio.AbstractEventLoop) -> None:
         self.base_url = config["gupshup.base_url"]
         self.read_url = config["gupshup.read_url"]
+        self.cloud_api_url = config["gupshup.cloud_api_url"]
         self.app_name = config["gupshup.app_name"]
         self.sender = config["gupshup.sender"]
         self.http = ClientSession(loop=loop)
@@ -153,3 +154,26 @@ async def send_location(
 
         response_data = json.loads(await resp.text())
         return {"status": resp.status, "messageId": response_data.get("messageId")}
+
+    async def send_reaction(self, message_id: str, emoji: str, type: str, data: dict):
+        """
+        Send a reaction to whatsapp
+
+        Parameters
+        ----------
+        message_id: str
+            The message ID of the reaction event
+        emoji: str
+            The emoji that was reacted with
+        type: str
+            The type of the reaction event
+        data: dict
+            The necessary data to send the reaction
+        """
+        headers = data.get("headers")
+        data.pop("headers")
+        data["message"] = json.dumps({"msgId": message_id, "type": type, "emoji": emoji})
+
+        resp = await self.http.post(self.cloud_api_url, data=data, headers=headers)
+        response_data = json.loads(await resp.text())
+        return response_data

gupshup_matrix/gupshup/data.py

@@ -47,6 +47,7 @@ class GupshupMessageData(SerializableAttrs):
     msg_gsId: str = attr.ib(default=None, metadata={"json": "gsId"})
     name: str = attr.ib(default=None, metadata={"json": "name"})
     address: str = attr.ib(default=None, metadata={"json": "address"})
+    emoji: str = attr.ib(default=None, metadata={"json": "emoji"})
 
 
 @dataclass

gupshup_matrix/gupshup/webhook.py

@@ -65,7 +65,6 @@ async def receive(self, request: web.Request) -> None:
                 f"Ignoring event because the gs_app [{data.get('app')}] is not registered."
             )
             return web.Response(status=406)
-
         if data.get("type") == GupshupEventType.MESSAGE:
             return await self.message_event(data)
         elif data.get("type") == GupshupEventType.MESSAGE_EVENT:
@@ -110,7 +109,10 @@ async def message_event(self, data: GupshupMessageEvent) -> web.Response:
         user: u.User = await u.User.get_by_gs_app(data.app)
         info = ChatInfo.deserialize(data.__dict__)
         info.sender = data.payload.sender
-        await portal.handle_gupshup_message(user, info, data)
+        if data.payload.type == "reaction":
+            await portal.handle_gupshup_reaction(user, data)
+        else:
+            await portal.handle_gupshup_message(user, info, data)
         return web.Response(status=204)
 
     async def status_event(self, data: GupshupStatusEvent) -> web.Response:

gupshup_matrix/matrix.py

@@ -1,13 +1,14 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Dict
 
 from mautrix.bridge import BaseMatrixHandler, RejectMatrixInvite
 from mautrix.types import (
     Event,
     EventID,
     EventType,
     ReactionEvent,
+    ReactionEventContent,
     RedactionEvent,
     RoomID,
     SingleReceiptEventContent,
@@ -47,9 +48,7 @@ async def handle_event(self, evt: Event) -> None:
             await self.handle_redaction(evt.room_id, evt.sender, evt.redacts, evt.event_id)
         elif evt.type == EventType.REACTION:
             evt: ReactionEvent
-            await self.handle_reaction(
-                evt.room_id, evt.sender, evt.event_id, evt.content, evt.timestamp
-            )
+            await self.handle_reaction(evt.room_id, evt.sender, evt.event_id, evt.content)
 
     async def handle_invite(
         self, room_id: RoomID, user_id: UserID, inviter: u.User, event_id: EventID
@@ -98,7 +97,7 @@ async def handle_redaction(
         if not portal:
             return
 
-        await portal.handle_matrix_redaction(user, event_id, redaction_event_id)
+        await portal.handle_matrix_redaction(user, event_id)
 
     async def allow_message(self, user: u.User) -> bool:
         return user.relay_whitelisted
@@ -111,3 +110,42 @@ async def handle_read_receipt(
     ) -> None:
         self.log.debug(f"Got read receipt for {event_id} from {user.mxid}")
         await portal.handle_matrix_read_receipt(event_id)
+
+    async def handle_reaction(
+        self,
+        room_id: RoomID,
+        user_id: UserID,
+        event_id: EventID,
+        content: ReactionEventContent,
+    ) -> None:
+        """
+        Send a reaction to a user in a room.
+
+        Parameters
+        ----------
+        room_id: RoomID
+            The room ID of the room where the reaction was sent
+        user_id: UserID
+            The user ID of the user who sent the reaction
+        event_id: EventID
+            The event ID of the reaction event
+        content: ReactionEventContent
+            The content of the reaction event
+        """
+        self.log.debug(f"Received reaction event: {content}")
+        user: u.User = await u.User.get_by_mxid(user_id)
+        message_mxid = content.relates_to.event_id
+        if not user:
+            return
+
+        if not message_mxid:
+            return
+
+        portal: po.Portal = await po.Portal.get_by_mxid(room_id)
+        if not portal:
+            return
+        try:
+            await portal.handle_matrix_reaction(user, message_mxid, event_id, room_id, content)
+        except ValueError as error:
+            self.log.error(f"Error trying to send a reaction {error}")
+            return