Use MLS member table indexed by group id (#2859)

wireapp · Nov 25, 2022 · 54b2cae · 54b2cae
1 parent 74f86a6
commit 54b2cae
Show file tree

Hide file tree

Showing 17 changed files with 295 additions and 113 deletions.
diff --git a/Makefile b/Makefile
@@ -211,7 +211,10 @@ git-add-cassandra-schema: db-reset git-add-cassandra-schema-impl
 .PHONY: git-add-cassandra-schema-impl
 git-add-cassandra-schema-impl:
 	$(eval CASSANDRA_CONTAINER := $(shell docker ps | grep '/cassandra:' | perl -ne '/^(\S+)\s/ && print $$1'))
-	( echo '-- automatically generated with `make git-add-cassandra-schema`' ; docker exec -i $(CASSANDRA_CONTAINER) /usr/bin/cqlsh -e "DESCRIBE schema;" ) > ./cassandra-schema.cql
+	( echo '-- automatically generated with `make git-add-cassandra-schema`'; \
+      docker exec -i $(CASSANDRA_CONTAINER) /usr/bin/cqlsh -e "DESCRIBE schema;" ) \
+    | sed "s/CREATE TABLE galley_test.member_client/-- NOTE: this table is unused. It was replaced by mls_group_member_client\nCREATE TABLE galley_test.member_client/g" \
+      > ./cassandra-schema.cql
 	git add ./cassandra-schema.cql
 
 .PHONY: cqlsh

diff --git a/cassandra-schema.cql b/cassandra-schema.cql
@@ -364,6 +364,7 @@ CREATE TABLE galley_test.group_id_conv_id (
     AND read_repair_chance = 0.0
     AND speculative_retry = '99PERCENTILE';
 
+-- NOTE: this table is unused. It was replaced by mls_group_member_client
 CREATE TABLE galley_test.member_client (
     conv uuid,
     user_domain text,
@@ -430,6 +431,29 @@ CREATE TABLE galley_test.conversation_codes (
     AND read_repair_chance = 0.0
     AND speculative_retry = '99PERCENTILE';
 
+CREATE TABLE galley_test.mls_group_member_client (
+    group_id blob,
+    user_domain text,
+    user uuid,
+    client text,
+    key_package_ref blob,
+    PRIMARY KEY (group_id, user_domain, user, client)
+) WITH CLUSTERING ORDER BY (user_domain ASC, user ASC, client ASC)
+    AND bloom_filter_fp_chance = 0.01
+    AND caching = {'keys': 'ALL', 'rows_per_partition': 'NONE'}
+    AND comment = ''
+    AND compaction = {'class': 'org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy', 'max_threshold': '32', 'min_threshold': '4'}
+    AND compression = {'chunk_length_in_kb': '64', 'class': 'org.apache.cassandra.io.compress.LZ4Compressor'}
+    AND crc_check_chance = 1.0
+    AND dclocal_read_repair_chance = 0.1
+    AND default_time_to_live = 0
+    AND gc_grace_seconds = 864000
+    AND max_index_interval = 2048
+    AND memtable_flush_period_in_ms = 0
+    AND min_index_interval = 128
+    AND read_repair_chance = 0.0
+    AND speculative_retry = '99PERCENTILE';
+
 CREATE TABLE galley_test.clients (
     user uuid PRIMARY KEY,
     clients set<text>

diff --git a/changelog.d/0-release-notes/member_clients_migration b/changelog.d/0-release-notes/member_clients_migration
@@ -0,0 +1 @@
+This realease migrates data from `galley.member_client` to `galley.mls_group_member_client`. When upgrading wire-server no manual steps are required.
diff --git a/services/galley/galley.cabal b/services/galley/galley.cabal
@@ -538,11 +538,15 @@ executable galley-integration
 
 executable galley-migrate-data
   main-is:            Main.hs
+
+  -- cabal-fmt: expand migrate-data/src
   other-modules:
     Galley.DataMigration
     Galley.DataMigration.Types
+    Main
     Paths_galley
     V1_BackfillBillingTeamMembers
+    V2_MigrateMLSMembers
 
   hs-source-dirs:     migrate-data/src
   default-extensions:
@@ -598,6 +602,7 @@ executable galley-migrate-data
     , exceptions
     , extended
     , extra                 >=1.3
+    , galley
     , galley-types
     , imports
     , lens
@@ -685,6 +690,7 @@ executable galley-schema
     V74_ExposeInvitationsToTeamAdmin
     V75_MLSGroupInfo
     V76_ProposalOrigin
+    V77_MLSGroupMemberClient
 
   hs-source-dirs:     schema/src
   default-extensions:

diff --git a/services/galley/migrate-data/src/Galley/DataMigration/Types.hs b/services/galley/migrate-data/src/Galley/DataMigration/Types.hs
@@ -42,7 +42,8 @@ newtype MigrationActionT m a = MigrationActionT {unMigrationAction :: ReaderT En
       Monad,
       MonadIO,
       MonadThrow,
-      MonadReader Env
+      MonadReader Env,
+      MonadUnliftIO
     )
 
 instance MonadTrans MigrationActionT where

diff --git a/services/galley/migrate-data/src/Main.hs b/services/galley/migrate-data/src/Main.hs
@@ -22,11 +22,17 @@ import Imports
 import Options.Applicative
 import qualified System.Logger.Extended as Log
 import qualified V1_BackfillBillingTeamMembers
+import qualified V2_MigrateMLSMembers
 
 main :: IO ()
 main = do
   o <- execParser (info (helper <*> cassandraSettingsParser) desc)
   l <- Log.mkLogger Log.Debug Nothing Nothing
-  migrate l o [V1_BackfillBillingTeamMembers.migration]
+  migrate
+    l
+    o
+    [ V1_BackfillBillingTeamMembers.migration,
+      V2_MigrateMLSMembers.migration
+    ]
   where
     desc = header "Galley Cassandra Data Migrations" <> fullDesc
diff --git a/services/galley/migrate-data/src/V2_MigrateMLSMembers.hs b/services/galley/migrate-data/src/V2_MigrateMLSMembers.hs
@@ -0,0 +1,101 @@
+-- This file is part of the Wire Server implementation.
+--
+-- Copyright (C) 2022 Wire Swiss GmbH <opensource@wire.com>
+--
+-- 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/>.
+
+module V2_MigrateMLSMembers where
+
+import Cassandra
+import Conduit
+import Data.Conduit.Internal (zipSources)
+import qualified Data.Conduit.List as C
+import Data.Domain
+import Data.Id
+import Data.Map.Strict (lookup)
+import qualified Data.Map.Strict as Map
+import Galley.Cassandra.Instances ()
+import Galley.DataMigration.Types
+import Imports hiding (lookup)
+import qualified System.Logger.Class as Log
+import UnliftIO (pooledMapConcurrentlyN_)
+import UnliftIO.Async (pooledMapConcurrentlyN)
+import Wire.API.MLS.Group
+import Wire.API.MLS.KeyPackage
+
+migration :: Migration
+migration =
+  Migration
+    { version = MigrationVersion 2,
+      text = "Migrating from member_client to mls_group_member_client",
+      action =
+        runConduit $
+          zipSources
+            (C.sourceList [(1 :: Int32) ..])
+            getMemberClientsFromLegacy
+            .| C.mapM_
+              ( \(i, rows) -> do
+                  Log.info (Log.field "Entries " (show (i * pageSize)))
+                  let convIds = map rowConvId rows
+                  m <- lookupGroupIds convIds
+                  let newRows = flip mapMaybe rows $ \(conv, domain, uid, client, ref) ->
+                        conv `lookup` m >>= \groupId -> pure (groupId, domain, uid, client, ref)
+                  insertMemberClients newRows
+              )
+    }
+
+rowConvId :: (ConvId, Domain, UserId, ClientId, KeyPackageRef) -> ConvId
+rowConvId (conv, _, _, _, _) = conv
+
+pageSize :: Int32
+pageSize = 1000
+
+getMemberClientsFromLegacy :: MonadClient m => ConduitM () [(ConvId, Domain, UserId, ClientId, KeyPackageRef)] m ()
+getMemberClientsFromLegacy = paginateC cql (paramsP LocalQuorum () pageSize) x5
+  where
+    cql :: PrepQuery R () (ConvId, Domain, UserId, ClientId, KeyPackageRef)
+    cql = "SELECT conv, user_domain, user, client, key_package_ref from member_client"
+
+lookupGroupIds :: [ConvId] -> MigrationActionT IO (Map ConvId GroupId)
+lookupGroupIds convIds = do
+  rows <- pooledMapConcurrentlyN 8 (\convId -> retry x5 (query1 cql (params LocalQuorum (Identity convId)))) convIds
+  rows' <-
+    rows
+      & mapM
+        ( \case
+            (Just (c, mg)) -> do
+              case mg of
+                Nothing -> do
+                  Log.warn (Log.msg ("No group found for conv " <> show c))
+                  pure Nothing
+                Just g -> pure (Just (c, g))
+            Nothing -> do
+              Log.warn (Log.msg ("Conversation is missing for entry" :: Text))
+              pure Nothing
+        )
+
+  rows'
+    & catMaybes
+    & Map.fromList
+    & pure
+  where
+    cql :: PrepQuery R (Identity ConvId) (ConvId, Maybe GroupId)
+    cql = "SELECT conv, group_id from conversation where conv = ?"
+
+insertMemberClients :: (MonadUnliftIO m, MonadClient m) => [(GroupId, Domain, UserId, ClientId, KeyPackageRef)] -> m ()
+insertMemberClients rows = do
+  pooledMapConcurrentlyN_ 8 (\row -> retry x5 (write cql (params LocalQuorum row))) rows
+  where
+    cql :: PrepQuery W (GroupId, Domain, UserId, ClientId, KeyPackageRef) ()
+    cql = "INSERT INTO mls_group_member_client (group_id, user_domain, user, client, key_package_ref) VALUES (?, ?, ?, ?, ?)"
diff --git a/services/galley/schema/src/Main.hs b/services/galley/schema/src/Main.hs
@@ -79,6 +79,7 @@ import qualified V73_MemberClientTable
 import qualified V74_ExposeInvitationsToTeamAdmin
 import qualified V75_MLSGroupInfo
 import qualified V76_ProposalOrigin
+import qualified V77_MLSGroupMemberClient
 
 main :: IO ()
 main = do
@@ -143,7 +144,8 @@ main = do
       V73_MemberClientTable.migration,
       V74_ExposeInvitationsToTeamAdmin.migration,
       V75_MLSGroupInfo.migration,
-      V76_ProposalOrigin.migration
+      V76_ProposalOrigin.migration,
+      V77_MLSGroupMemberClient.migration
       -- When adding migrations here, don't forget to update
       -- 'schemaVersion' in Galley.Cassandra
       -- (see also docs/developer/cassandra-interaction.md)

diff --git a/services/galley/schema/src/V77_MLSGroupMemberClient.hs b/services/galley/schema/src/V77_MLSGroupMemberClient.hs
@@ -0,0 +1,36 @@
+-- This file is part of the Wire Server implementation.
+--
+-- Copyright (C) 2022 Wire Swiss GmbH <opensource@wire.com>
+--
+-- 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/>.
+
+module V77_MLSGroupMemberClient (migration) where
+
+import Cassandra.Schema
+import Imports
+import Text.RawString.QQ
+
+migration :: Migration
+migration =
+  Migration 77 "Add table mls_group_member_client which replaces member_client" $ do
+    schema'
+      [r| CREATE TABLE mls_group_member_client (
+            group_id blob,
+            user_domain text,
+            user uuid,
+            client text,
+            key_package_ref blob,
+            PRIMARY KEY (group_id, user_domain, user, client)
+          );
+        |]
diff --git a/services/galley/src/Galley/API/Create.hs b/services/galley/src/Galley/API/Create.hs
@@ -126,11 +126,11 @@ createGroupConversation lusr conn newConv = do
   conv <- E.createConversation lcnv nc
 
   -- set creator client for MLS conversations
-  case (newConvProtocol newConv, newConvCreatorClient newConv) of
-    (ProtocolProteusTag, _) -> pure ()
-    (ProtocolMLSTag, Just c) ->
-      E.addMLSClients lcnv (qUntagged lusr) (Set.singleton (c, nullKeyPackageRef))
-    (ProtocolMLSTag, Nothing) ->
+  case (convProtocol conv, newConvCreatorClient newConv) of
+    (ProtocolProteus, _) -> pure ()
+    (ProtocolMLS mlsMeta, Just c) ->
+      E.addMLSClients (cnvmlsGroupId mlsMeta) (qUntagged lusr) (Set.singleton (c, nullKeyPackageRef))
+    (ProtocolMLS _mlsMeta, Nothing) ->
       throw (InvalidPayload "Missing creator_client field when creating an MLS conversation")
 
   now <- input