add raw implementations for gdm functions

- pubsub: add mass_sub() method
 - gateway.websocket: use mass_sub()
 - storage: add last_message_id to group dm channels
 - user_storage: add get_gdms_internal()

litecord/blueprints/dm_channels.py

@@ -30,6 +30,20 @@ log = Logger(__name__)
 bp = Blueprint('dm_channels', __name__)
 
 
+async def _raw_gdm_add(channel_id, user_id):
+    await app.db.execute("""
+    INSERT INTO group_dm_members (id, member_id)
+    VALUES ($1, $2)
+    """, channel_id, user_id)
+
+
+async def _raw_gdm_remove(channel_id, user_id):
+    await app.db.execute("""
+    DELETE FROM group_dm_members
+    WHERE id = $1 AND member_id = $2
+    """, channel_id, user_id)
+
+
 async def _gdm_create(user_id, peer_id) -> int:
     """Create a group dm, given two users.
 
@@ -37,7 +51,24 @@ async def _gdm_create(user_id, peer_id) -> int:
     """
     channel_id = get_snowflake()
 
-    # TODO
+    await app.db.execute("""
+    INSERT INTO channels (id, channel_type)
+    VALUES ($1, $2)
+    """, channel_id, ChannelType.GROUP_DM.value)
+
+    await app.db.execute("""
+    INSERT INTO group_dm_channels (id, owner_id, name, icon)
+    VALUES ($1, $2, NULL, NULL)
+    """, channel_id, user_id)
+
+    await _raw_gdm_add(channel_id, user_id)
+    await _raw_gdm_add(channel_id, peer_id)
+
+    await app.dispatcher.sub('channel', channel_id, user_id)
+    await app.dispatcher.sub('channel', channel_id, peer_id)
+
+    chan = await app.storage.get_channel(channel_id)
+    await app.dispatcher.dispatch('channel', 'CHANNEL_CREATE', chan)
 
     return channel_id
 
@@ -50,9 +81,11 @@ async def _gdm_add_recipient(channel_id: int, peer_id: int, *, user_id=None):
      - A CHANNEL_CREATE to the peer.
      - A CHANNEL_UPDATE to all remaining recipients.
     """
+    await _raw_gdm_add(channel_id, peer_id)
 
-    # TODO
-    pass
+    chan = await app.storage.get_channel(channel_id)
+    await app.dispatcher.dispatch('user', user_id, 'CHANNEL_CREATE', chan)
+    await app.dispatcher.dispatch('channel', channel_id, 'CHANNEL_UPDATE', chan)
 
 
 async def _gdm_remove_recipient(channel_id: int, peer_id: int, *, user_id=None):
@@ -64,9 +97,11 @@ async def _gdm_remove_recipient(channel_id: int, peer_id: int, *, user_id=None):
      - A CHANNEL_DELETE to the peer.
      - A CHANNEL_UPDATE to all remaining recipients.
     """
+    await _raw_gdm_remove(channel_id, peer_id)
 
-    # TODO
-    pass
+    chan = await app.storage.get_channel(channel_id)
+    await app.dispatcher.dispatch('user', user_id, 'CHANNEL_DELETE', chan)
+    await app.dispatcher.dispatch('channel', channel_id, 'CHANNEL_UPDATE', chan)
 
 
 @bp.route('/<int:dm_chan>/receipients/<int:peer_id>', methods=['PUT'])

litecord/dispatcher.py

@@ -103,6 +103,15 @@ class EventDispatcher:
         for key in keys:
             await self.subscribe(backend_str, key, identifier)
 
+    async def mass_sub(self, identifier: Any,
+                       backends: List[tuple]):
+        """Mass subscribe to many backends at once."""
+        for backend_str, keys in backends:
+            log.debug('subscribing {} to {} keys in backend {}',
+                      identifier, len(keys), backend_str)
+
+            await self.sub_many(backend_str, identifier, keys)
+
     async def dispatch(self, backend_str: str, key: Any, *args, **kwargs):
         """Dispatch an event to the backend.
 

litecord/gateway/websocket.py

@@ -437,17 +437,24 @@ class GatewayWebsocket:
             by GuildDispatcher.sub
         """
         user_id = self.state.user_id
-
         guild_ids = await self._guild_ids()
-        log.info('subscribing to {} guilds', len(guild_ids))
-        await self.ext.dispatcher.sub_many('guild', user_id, guild_ids)
 
         # subscribe the user to all dms they have OPENED.
         dms = await self.user_storage.get_dms(user_id)
         dm_ids = [int(dm['id']) for dm in dms]
 
+        # fetch all group dms the user is a member of.
+        gdm_ids = await self.user_storage.get_gdms_internal(user_id)
+
+        log.info('subscribing to {} guilds', len(guild_ids))
         log.info('subscribing to {} dms', len(dm_ids))
-        await self.ext.dispatcher.sub_many('channel', user_id, dm_ids)
+        log.info('subscribing to {} group dms', len(gdm_ids))
+
+        await self.ext.dispatcher.mass_sub(user_id, [
+            ('guild', guild_ids),
+            ('channel', dm_ids),
+            ('channel', gdm_ids)
+        ])
 
         if not self.state.bot:
             # subscribe to all friends

litecord/storage.py

@@ -414,7 +414,8 @@ class Storage:
             drow['type'] = chan_type
 
             drow['last_message_id'] = await self.chan_last_message_str(
-                channel_id)
+                channel_id
+            )
 
             # dms have just two recipients.
             drow['recipients'] = [
@@ -436,6 +437,9 @@ class Storage:
 
             drow = dict(gdm_row)
             drow['recipients'] = await self._gdm_recipients(channel_id)
+            drow['last_message_id'] = await self.chan_last_message_str(
+                channel_id
+            )
 
             return drow
 

litecord/user_storage.py

@@ -338,3 +338,13 @@ class UserStorage:
                 )
             )
         """, user_id, peer_id)
+
+    async def get_gdms_internal(self, user_id) -> List[int]:
+        """Return a list of Group DM IDs the user is a member of."""
+        rows = await self.db.fetch("""
+        SELECT id
+        FROM group_dm_members
+        WHERE member_id = $1
+        """, user_id)
+
+        return [r['id'] for r in rows]