diff --git a/cassandra-schema.cql b/cassandra-schema.cql
index 9e5e1528992..160a310f39b 100644
--- a/cassandra-schema.cql
+++ b/cassandra-schema.cql
@@ -738,7 +738,7 @@ CREATE TABLE brig_test.rich_info (
AND crc_check_chance = 1.0
AND default_time_to_live = 0
AND extensions = {}
- AND gc_grace_seconds = 864000
+ AND gc_grace_seconds = 86400
AND max_index_interval = 2048
AND memtable_flush_period_in_ms = 0
AND min_index_interval = 128
@@ -848,7 +848,7 @@ CREATE TABLE brig_test.service_team (
AND crc_check_chance = 1.0
AND default_time_to_live = 0
AND extensions = {}
- AND gc_grace_seconds = 864000
+ AND gc_grace_seconds = 86400
AND max_index_interval = 2048
AND memtable_flush_period_in_ms = 0
AND min_index_interval = 128
@@ -874,7 +874,7 @@ CREATE TABLE brig_test.service_user (
AND crc_check_chance = 1.0
AND default_time_to_live = 0
AND extensions = {}
- AND gc_grace_seconds = 864000
+ AND gc_grace_seconds = 86400
AND max_index_interval = 2048
AND memtable_flush_period_in_ms = 0
AND min_index_interval = 128
@@ -1088,7 +1088,7 @@ CREATE TABLE brig_test.user (
AND crc_check_chance = 1.0
AND default_time_to_live = 0
AND extensions = {}
- AND gc_grace_seconds = 864000
+ AND gc_grace_seconds = 86400
AND max_index_interval = 2048
AND memtable_flush_period_in_ms = 0
AND min_index_interval = 128
@@ -1137,7 +1137,7 @@ CREATE TABLE brig_test.user_handle (
AND crc_check_chance = 1.0
AND default_time_to_live = 0
AND extensions = {}
- AND gc_grace_seconds = 864000
+ AND gc_grace_seconds = 86400
AND max_index_interval = 2048
AND memtable_flush_period_in_ms = 0
AND min_index_interval = 128
diff --git a/changelog.d/2-features/user-pg-migration b/changelog.d/2-features/user-pg-migration
new file mode 100644
index 00000000000..2abecf732fc
--- /dev/null
+++ b/changelog.d/2-features/user-pg-migration
@@ -0,0 +1 @@
+Support migrating user data to postgresql from cassandra
\ No newline at end of file
diff --git a/charts/wire-server/templates/background-worker/configmap.yaml b/charts/wire-server/templates/background-worker/configmap.yaml
index d4fe2a63202..299d0703d3a 100644
--- a/charts/wire-server/templates/background-worker/configmap.yaml
+++ b/charts/wire-server/templates/background-worker/configmap.yaml
@@ -84,6 +84,7 @@ data:
migrateConversationCodes: {{ .migrateConversationCodes }}
migrateTeamFeatures: {{ .migrateTeamFeatures }}
migrateDomainRegistration: {{ .migrateDomainRegistration }}
+ migrateUsers: {{ .migrateUsers }}
migrationOptions:
{{ toYaml .migrationOptions | indent 6 }}
diff --git a/charts/wire-server/values.yaml b/charts/wire-server/values.yaml
index 5bb0b276eb6..6af937084d5 100644
--- a/charts/wire-server/values.yaml
+++ b/charts/wire-server/values.yaml
@@ -1026,6 +1026,10 @@ background-worker:
# It's important to set `settings.postgresMigration.domainRegistration` to `migration-to-postgresql`
# before starting the migration.
migrateDomainRegistration: false
+ # This will start the migration of users
+ # It's important to set `settings.postgresMigration.users` to `migration-to-postgresql`
+ # before starting the migration.
+ migrateUsers: false
backendNotificationPusher:
pushBackoffMinWait: 10000 # in microseconds, so 10ms
diff --git a/deploy/dockerephemeral/docker-compose.yaml b/deploy/dockerephemeral/docker-compose.yaml
index 2081edddc21..fb2a4801ebc 100644
--- a/deploy/dockerephemeral/docker-compose.yaml
+++ b/deploy/dockerephemeral/docker-compose.yaml
@@ -290,7 +290,7 @@ services:
POSTGRES_PASSWORD: "posty-the-gres"
POSTGRES_USER: "wire-server"
POSTGRES_DB: "backendA"
- command: postgres -c max_connections=150
+ command: postgres -c max_connections=1000
cassandra:
container_name: demo_wire_cassandra
diff --git a/docs/src/developer/reference/config-options.md b/docs/src/developer/reference/config-options.md
index cf86ee7681c..570a4d355d1 100644
--- a/docs/src/developer/reference/config-options.md
+++ b/docs/src/developer/reference/config-options.md
@@ -2187,6 +2187,7 @@ The current settings and their background-worker flags are:
- `conversationCodes` -> `migrateConversationCodes`
- `teamFeatures` -> `migrateTeamFeatures`
- `domainRegistration` -> `migrateDomainRegistration`
+- `user` -> `migrateUsers`
**Migration pattern per migration setting**
@@ -2205,13 +2206,15 @@ The current settings and their background-worker flags are:
conversation: migration-to-postgresql
conversationCodes: migration-to-postgresql
teamFeatures: migration-to-postgresql
- domainRegistration: cassandra
+ domainRegistration: migration-to-postgresql
+ user: migration-to-postgresql
background-worker:
config:
migrateConversations: false
migrateConversationCodes: false
migrateTeamFeatures: false
migrateDomainRegistration: false
+ migrateUsers: false
```
This change should restart the affected pods, and new writes will follow the
@@ -2226,6 +2229,7 @@ The current settings and their background-worker flags are:
migrateConversationCodes: true
migrateTeamFeatures: true
migrateDomainRegistration: true
+ migrateUsers: true
```
During migration, Cassandra rows are not deleted. Writes and migration share
@@ -2241,6 +2245,16 @@ The current settings and their background-worker flags are:
- `conversationCodes`: `wire_conv_codes_migration_finished`
- `teamFeatures`: `wire_team_features_migration_finished`
- `domainRegistration`: `wire_domain_registration_migration_finished`
+ - `user`: `wire_user_migration_finished`
+
+ > ⚠️ For user migrations please watch the logs for `Invalid user found,
+ > skipping`. This would be accompanied by an error which is either
+ > `UserHasNoName` or `UserHasNoActivated`. These users are invalid and all
+ > interactions with them were resulting in errors. If these warnings are
+ > ignored, these users will stop existing in the system. If these users are
+ > to be saved, the operator must insert some value as `name` and/or
+ > `activated` and then re-trigger the migration **after** the background
+ > worker finishes migrating the valid users.
3. Cut over reads and writes to PostgreSQL for the selected migration
setting(s). This configuration must be used from now on for every new
@@ -2253,13 +2267,15 @@ The current settings and their background-worker flags are:
conversation: postgresql
conversationCodes: postgresql
teamFeatures: postgresql
- domainRegistration: cassandra
+ domainRegistration: postgresql
+ user: postgresql
background-worker:
config:
migrateConversations: false
migrateConversationCodes: false
migrateTeamFeatures: false
migrateDomainRegistration: false
+ migrateUsers: false
```
**How to run migrations independently or in batches**
diff --git a/integration/default.nix b/integration/default.nix
index 8290943dee2..a1ec576d096 100644
--- a/integration/default.nix
+++ b/integration/default.nix
@@ -57,6 +57,7 @@
, optparse-applicative
, process
, proto-lens
+, QuickCheck
, ram
, random
, raw-strings-qq
@@ -162,6 +163,7 @@ mkDerivation {
optparse-applicative
process
proto-lens
+ QuickCheck
ram
random
raw-strings-qq
diff --git a/integration/integration.cabal b/integration/integration.cabal
index d124256a118..8dd9466197c 100644
--- a/integration/integration.cabal
+++ b/integration/integration.cabal
@@ -181,6 +181,7 @@ library
Test.Migration.ConversationCodes
Test.Migration.DomainRegistration
Test.Migration.TeamFeatures
+ Test.Migration.User
Test.Migration.Util
Test.MLS
Test.MLS.Clients
@@ -301,6 +302,7 @@ library
, optparse-applicative
, process
, proto-lens
+ , QuickCheck ^>=2.15.0.1
, ram
, random
, raw-strings-qq
diff --git a/integration/test/API/Brig.hs b/integration/test/API/Brig.hs
index 7614567b84f..480bb781a15 100644
--- a/integration/test/API/Brig.hs
+++ b/integration/test/API/Brig.hs
@@ -156,10 +156,13 @@ getSelfClients u =
-- | https://staging-nginz-https.zinfra.io/v5/api/swagger-ui/#/default/delete_self
deleteUser :: (HasCallStack, MakesValue user) => user -> App Response
-deleteUser user = do
+deleteUser user = deleteUserWithPassword user (Just defPassword)
+
+deleteUserWithPassword :: (HasCallStack, MakesValue user) => user -> Maybe String -> App Response
+deleteUserWithPassword user mPassword = do
req <- baseRequest user Brig Versioned "/self"
submit "DELETE" $
- req & addJSONObject ["password" .= defPassword]
+ req & addJSONObject ["password" .= mPassword]
-- | https://staging-nginz-https.zinfra.io/v5/api/swagger-ui/#/default/post_clients
addClient ::
@@ -824,6 +827,15 @@ addBot user providerId serviceId convId = do
& zType "access"
& addJSONObject ["provider" .= providerId, "service" .= serviceId]
+rmBotSelf :: (HasCallStack, MakesValue domain) => domain -> String -> String -> App Response
+rmBotSelf domain bid cid = do
+ req <- rawBaseRequest domain Brig Versioned $ joinHttpPath ["bot", "self"]
+ submit "DELETE" $
+ req
+ & zType "bot"
+ & addHeader "Z-Bot" bid
+ & addHeader "Z-Conversation" cid
+
setProperty :: (MakesValue user, ToJSON val) => user -> String -> val -> App Response
setProperty user propName val = do
req <- baseRequest user Brig Versioned $ joinHttpPath ["properties", propName]
diff --git a/integration/test/API/Common.hs b/integration/test/API/Common.hs
index cd4b4b348ab..9b720a3b9bd 100644
--- a/integration/test/API/Common.hs
+++ b/integration/test/API/Common.hs
@@ -25,6 +25,7 @@ import qualified Data.ByteString as BS
import Data.Scientific (scientific)
import qualified Data.Vector as Vector
import System.Random (randomIO, randomRIO)
+import Test.QuickCheck
import Testlib.Prelude
-- | please don't use special shell characters like '!' here. it makes writing shell lines
@@ -33,8 +34,11 @@ defPassword :: String
defPassword = "hunter2."
randomEmail :: App String
-randomEmail = do
- u <- randomName
+randomEmail = liftIO $ generate arbitraryEmail
+
+arbitraryEmail :: Gen String
+arbitraryEmail = do
+ u <- arbitraryName
pure $ u <> "@example.com"
randomDomain :: App String
@@ -52,23 +56,32 @@ randomExternalId = liftIO $ do
pick = (chars !) <$> randomRIO (Array.bounds chars)
randomName :: App String
-randomName = liftIO $ do
- n <- randomRIO (8, 15)
+randomName = liftIO $ generate arbitraryName
+
+arbitraryName :: Gen String
+arbitraryName = do
+ n <- chooseInt (8, 15)
replicateM n pick
where
chars = mkArray $ ['A' .. 'Z'] <> ['a' .. 'z'] <> ['0' .. '9']
- pick = (chars !) <$> randomRIO (Array.bounds chars)
+ pick = (chars !) <$> chooseInt (Array.bounds chars)
randomHandle :: App String
-randomHandle = randomHandleWithRange 50 256
+randomHandle = liftIO $ generate arbitraryHandle
randomHandleWithRange :: Int -> Int -> App String
-randomHandleWithRange min' max' = liftIO $ do
- n <- randomRIO (min', max')
+randomHandleWithRange min' max' = liftIO $ generate (arbitraryHandleWithRange min' max')
+
+arbitraryHandle :: Gen String
+arbitraryHandle = arbitraryHandleWithRange 50 60
+
+arbitraryHandleWithRange :: Int -> Int -> Gen String
+arbitraryHandleWithRange min' max' = do
+ n <- chooseInt (min', max')
replicateM n pick
where
chars = mkArray $ ['a' .. 'z'] <> ['0' .. '9'] <> "_-."
- pick = (chars !) <$> randomRIO (Array.bounds chars)
+ pick = (chars !) <$> chooseInt (Array.bounds chars)
randomBytes :: Int -> App ByteString
randomBytes n = liftIO $ BS.pack <$> replicateM n randomIO
@@ -85,6 +98,14 @@ randomAlphaString n = liftIO $ replicateM n pick
chars = mkArray $ ['A' .. 'Z'] <> ['a' .. 'z'] <> ['0' .. '9']
pick = (chars !) <$> randomRIO (Array.bounds chars)
+randomPassword :: App String
+randomPassword = liftIO $ generate arbitraryPassword
+
+arbitraryPassword :: Gen String
+arbitraryPassword = do
+ n <- chooseInt (8, 1024)
+ replicateM n arbitraryPrintableChar
+
randomJSON :: App Value
randomJSON = do
let maxThings = 5
diff --git a/integration/test/SetupHelpers.hs b/integration/test/SetupHelpers.hs
index f4ed0509567..b5aed2c98a2 100644
--- a/integration/test/SetupHelpers.hs
+++ b/integration/test/SetupHelpers.hs
@@ -605,10 +605,14 @@ getCookieWithSamlLogin ::
App (Maybe String, SAML.SignedAuthnResponse)
getCookieWithSamlLogin mbZHost domain expectSuccess tid nameId mLabel (iid, (meta, privcreds)) = do
let idpConfig = SAML.IdPConfig (SAML.IdPId (fromMaybe (error "invalid idp id") (UUID.fromString iid))) meta ()
- spmeta <- getSPMetadataWithZHost domain mbZHost tid
- authnreq <- initiateSamlLoginWithZHostAndLabel domain mbZHost mLabel iid
- let spMetaData = toSPMetaData spmeta.body
- parsedAuthnReq = parseAuthnReqResp authnreq.body
+ spMetaData <-
+ getSPMetadataWithZHost domain mbZHost tid `bindResponse` \resp -> do
+ resp.status `shouldMatchInt` 200
+ pure $ toSPMetaData resp.body
+ parsedAuthnReq <-
+ initiateSamlLoginWithZHostAndLabel domain mbZHost mLabel iid `bindResponse` \resp -> do
+ resp.status `shouldMatchInt` 200
+ pure $ parseAuthnReqResp resp.body
authnReqResp <- makeAuthnResponse nameId privcreds idpConfig spMetaData parsedAuthnReq
mCookie <- finalizeSamlLoginWithZHost domain mbZHost tid authnReqResp `bindResponse` validateLoginResp
pure (mCookie, authnReqResp)
@@ -653,13 +657,14 @@ makeAuthnResponse nameId privcreds idpConfig spMetaData parsedAuthnReq =
-- | extract an `AuthnRequest` from the html form in the http response from /sso/initiate-login
parseAuthnReqResp ::
+ (HasCallStack) =>
ByteString ->
SAML.AuthnRequest
parseAuthnReqResp bs = reqBody
where
xml :: XML.Document
xml =
- fromRight (error "malformed html in response body") $
+ fromRight (error $ "malformed html in response body: \n" <> show bs) $
XML.parseText XML.def (cs bs)
reqBody :: SAML.AuthnRequest
@@ -855,12 +860,12 @@ createNewIndex = do
ExitFailure _ -> assertFailure $ prefix <> "failed to create index"
ExitSuccess -> pure indexName
-reindexUsers :: (HasCallStack) => BackendResource -> Int -> App ()
-reindexUsers ber pageSize = do
+reindexUsers :: (HasCallStack) => BackendResource -> ServiceOverrides -> Int -> App ()
+reindexUsers ber serviceOverrides pageSize = do
testName <- asks (fromMaybe "NoTest" . (.currentTestName))
let indexName = ber.berElasticsearchIndex
let prefix = "[reindex-users:" <> indexName <> ":" <> testName <> "] "
- getBrigConfig <- readAndUpdateConfig (defaultOverrides ber) ber Brig
+ getBrigConfig <- readAndUpdateConfig (defaultOverrides ber <> serviceOverrides) ber Brig
brigConfig <- liftIO $ getBrigConfig
esServer <- brigConfig %. "elasticsearch.url" & asString
esCredentials <- brigConfig %. "elasticsearch.credentials" & asString
diff --git a/integration/test/Test/Migration/Conversation.hs b/integration/test/Test/Migration/Conversation.hs
index 88f78e059ea..1e712ba9094 100644
--- a/integration/test/Test/Migration/Conversation.hs
+++ b/integration/test/Test/Migration/Conversation.hs
@@ -99,6 +99,7 @@ testMigrationToPostgresMLS = do
runPhase 5
where
n = 1
+
createTestConvs :: (HasCallStack) => ClientIdentity -> String -> ClientIdentity -> ClientIdentity -> [ClientIdentity] -> App TestConvList
createTestConvs creatorC tid melC markC othersC = do
unmodifiedConvs <- replicateM n $ do
diff --git a/integration/test/Test/Migration/User.hs b/integration/test/Test/Migration/User.hs
new file mode 100644
index 00000000000..573b3f46e7c
--- /dev/null
+++ b/integration/test/Test/Migration/User.hs
@@ -0,0 +1,956 @@
+{-# LANGUAGE ApplicativeDo #-}
+{-# OPTIONS_GHC -Wno-ambiguous-fields #-}
+
+-- | The migration has these phases.
+-- 1. Write to cassandra (before any migration activity)
+-- 2. Galley is prepared for migrations (new things created in PG, old things are in Cassandra)
+-- 3. Backgound worker starts migration
+-- 4. Background worker finishes migration, galley is still configured to think migration is on going
+-- 5. Background worker is configured to not do anything, galley is configured to only use PG
+--
+-- The comments and variable names call these phases by number i.e. Phase1, Phase2, and so on.
+--
+-- The tests are from the perspective of mel, a user on the dynamic backend,
+-- called backendM (migrating backend). There are also users called mark and mia
+-- on this backend.
+module Test.Migration.User where
+
+import API.Brig
+import qualified API.BrigInternal as I
+import API.Common
+import API.Galley
+import qualified API.GalleyInternal as I
+import API.Spar
+import Control.Applicative
+import Control.Monad.Codensity
+import Control.Monad.Reader
+import qualified Data.Aeson.KeyMap as KM
+import qualified Data.Aeson.KeyMap as KeyMap
+import Data.IntMap (IntMap)
+import qualified Data.IntMap as IntMap
+import qualified Data.IntSet as IntSet
+import qualified Data.Map as Map
+import Data.String.Conversions
+import Data.Text (Text)
+import qualified Data.Text as Text
+import Data.Tuple.Extra
+import Data.UUID (UUID)
+import qualified Data.UUID as UUID
+import qualified Data.Vector as Vector
+import Database.CQL.IO
+import GHC.Stack
+import Notifications
+import SetupHelpers hiding (deleteUser)
+import Test.Bot (mkBotService)
+import Test.Migration.Util
+import Test.QuickCheck
+import Test.Search
+import Testlib.MockIntegrationService (MockServerSettings (..), withMockServer)
+import Testlib.Prelude
+import Testlib.ResourcePool
+import UnliftIO
+
+testUserMigrationToPostgres :: App ()
+testUserMigrationToPostgres = withMockServer botServiceSettings mkBotService $ \(botHost, botPort) _botChan -> do
+ resourcePool <- asks (.resourcePool)
+
+ runCodensity (acquireResources 1 resourcePool) $ \[migratingBackend] -> do
+ let domainM = migratingBackend.berDomain
+ (mel, pid, sid, seedUsers) <- runCodensity (startDynamicBackend migratingBackend phase1Overrides) $ \_ -> do
+ -- mel exists to connect with all the personal users, so we can wait for a
+ -- notification for their deletion
+ mel <- randomUser domainM def
+
+ pid <- setupProvider domainM def {newProviderPassword = Just defPassword} %. "id" & asString
+ service <-
+ newService domainM pid
+ $ def
+ { newServiceUrl = "https://" <> botHost <> ":" <> show botPort,
+ newServiceKey = cs botServiceSettings.publicKey
+ }
+ sid <- service %. "id" & asString
+ updateServiceConn domainM pid sid (object ["password" .= defPassword, "enabled" .= True]) >>= assertSuccess
+
+ seedUsers <- seedTestUsers domainM mel pid sid
+ pure (mel, pid, sid, seedUsers)
+
+ newUsersRef <- newIORef mempty
+ updatedUsersRef <- newIORef mempty
+ updates <- fmap IntMap.fromList . for [1 .. 5] $ \phase -> do
+ (phase,) <$> liftIO (generate (arbitraryPhaseUpdates nUpdates))
+
+ addUsersToFailureContext [("mel", mel)]
+ $ addJSONToFailureContext "updates" updates
+ $ addJSONToFailureContext "seed users" seedUsers do
+ let runPhase :: (HasCallStack) => Int -> App ()
+ runPhase phase = do
+ runCodensity (startDynamicBackend migratingBackend (phaseOverrides IntMap.! phase)) $ \_ -> do
+ let toBeUpdated = seedUsers.updates IntMap.! phase
+ phaseUpdates = updates IntMap.! phase
+
+ updatedScimUsersWithRichInfo <- updateScimUsers domainM toBeUpdated.scimUsersWithRichInfo phaseUpdates.scimUsersWithRichInfo
+ updatedScimUsersWithoutRichInfo <- updateScimUsers domainM toBeUpdated.scimUsersWithoutRichInfo phaseUpdates.scimUsersWithoutRichInfo
+ updatedPendingScimUsers <- updatePendingScimUsers domainM toBeUpdated.pendingScimUsers phaseUpdates.pendingScimUsers
+ updatedSsoUsers <- checkUpdateUser toBeUpdated.ssoUsers.users phaseUpdates.ssoUsers
+ updatedPasswordTeamUsers <- checkUpdateUser toBeUpdated.passwordTeamUsers.users phaseUpdates.passwordTeamUsers
+ updatedPersonalUsersWithoutHandle <- checkUpdateUser toBeUpdated.personalUsersWithoutHandle phaseUpdates.personalUsersWithoutHandle
+ updatedPersonalUsersWithHandle <- checkUpdateUser toBeUpdated.personalUsersWithHandle phaseUpdates.personalUsersWithHandle
+ let updatedUsers =
+ TestUserList
+ { scimUsersWithRichInfo = updatedScimUsersWithRichInfo,
+ scimUsersWithoutRichInfo = updatedScimUsersWithoutRichInfo,
+ pendingScimUsers = updatedPendingScimUsers,
+ ssoUsers = toBeUpdated.ssoUsers {users = updatedSsoUsers} :: TestTeamUsers,
+ passwordTeamUsers = toBeUpdated.passwordTeamUsers {users = updatedPasswordTeamUsers} :: TestTeamUsers,
+ personalUsersWithoutHandle = updatedPersonalUsersWithoutHandle,
+ personalUsersWithHandle = updatedPersonalUsersWithHandle,
+ -- Bots don't have any updates
+ botsInTeamConvs = toBeUpdated.botsInTeamConvs,
+ botsInPersonalConvs = toBeUpdated.botsInPersonalConvs
+ }
+
+ newUsers <- createTestUsers domainM mel pid sid nNew
+ modifyIORef newUsersRef (IntMap.insert phase newUsers)
+ modifyIORef updatedUsersRef (IntMap.insert phase updatedUsers)
+
+ let toBeDeleted = seedUsers.deletes IntMap.! phase
+
+ deleteScimUsers domainM False toBeDeleted.scimUsersWithRichInfo
+ deleteScimUsers domainM False toBeDeleted.scimUsersWithoutRichInfo
+ deleteScimUsers domainM True toBeDeleted.pendingScimUsers
+ deleteTeamUsers toBeDeleted.ssoUsers
+ deleteTeamUsers toBeDeleted.passwordTeamUsers
+ deletePersonalUsers mel toBeDeleted.personalUsersWithoutHandle
+ deletePersonalUsers mel toBeDeleted.personalUsersWithHandle
+ deleteBotsTeam toBeDeleted.botsInTeamConvs pid sid
+ deleteBotConvs mel toBeDeleted.botsInPersonalConvs
+
+ checkAllDeletionsWorked domainM mel seedUsers.deletes phase
+ checkUnaffectedUsers domainM seedUsers.deletes seedUsers.updates phase
+ updatedSoFar <- readIORef updatedUsersRef
+ newSoFar <- readIORef newUsersRef
+ addJSONToFailureContext "newSoFar" newSoFar
+ $ checkNewAndUpdatedUsers domainM updatedSoFar newSoFar
+
+ when (phase == 3) $ do
+ waitForMigration domainM userMigrationFinishedCounterName
+ runPhase 1
+ runPhase 2
+ runPhase 3
+ runPhase 4
+ runPhase 5
+ where
+ parallelism = 64
+
+ -- Number of users of each type
+ nUpdates = 5
+ nDeletes = 1
+ nNew = 1
+
+ botServiceSettings = def
+
+ seedTestUsers :: (HasCallStack, MakesValue mel) => String -> mel -> String -> String -> App TestUsersByOperations
+ seedTestUsers domain mel pid sid =
+ fmap mconcat . for [(1 :: Int) .. 5] $ \phase -> do
+ updates <- IntMap.singleton phase <$> createTestUsers domain mel pid sid nUpdates
+ deletes <- IntMap.singleton phase <$> createTestUsers domain mel pid sid nDeletes
+ pure TestUsersByOperations {..}
+
+ tombstone :: String -> String -> Maybe String -> Value
+ tombstone domain uid mTid =
+ object
+ $ [ "accent_id" .= (0 :: Int),
+ "assets" .= (),
+ "deleted" .= True,
+ "id" .= uid,
+ "legalhold_status" .= "no_consent",
+ "name" .= "default",
+ "picture" .= (),
+ "qualified_id" .= object ["domain" .= domain, "id" .= uid],
+ "searchable" .= True,
+ "supported_protocols" .= ["proteus"],
+ "type" .= "regular"
+ ]
+ <> (maybe [] (\tid -> ["team" .= tid]) mTid)
+
+ scimUserIdsWithGetter :: (HasCallStack) => IntMap TestUserList -> [(String, String)]
+ scimUserIdsWithGetter relevantSeedUsers =
+ foldMap IntMap.elems . for relevantSeedUsers $ \usersInPhase -> do
+ map (usersInPhase.scimUsersWithRichInfo.token,) (Map.keys usersInPhase.scimUsersWithRichInfo.users)
+ <> map (usersInPhase.scimUsersWithoutRichInfo.token,) (Map.keys usersInPhase.scimUsersWithoutRichInfo.users)
+ <> map (usersInPhase.pendingScimUsers.token,) (Map.keys usersInPhase.pendingScimUsers.users)
+
+ nonScimUserIds :: (HasCallStack) => Value -> IntMap TestUserList -> [(Value, Value)]
+ nonScimUserIds mel relevantSeedUsers = foldMap IntMap.elems . for relevantSeedUsers $ \usersInPhase -> do
+ map (usersInPhase.scimUsersWithRichInfo.owner,) (thd3 <$> Map.elems usersInPhase.scimUsersWithRichInfo.users)
+ <> map (usersInPhase.scimUsersWithoutRichInfo.owner,) (thd3 <$> Map.elems usersInPhase.scimUsersWithoutRichInfo.users)
+ <> map (usersInPhase.passwordTeamUsers.owner,) (fst <$> Map.elems usersInPhase.passwordTeamUsers.users)
+ <> map (mel,) (fst <$> Map.elems usersInPhase.personalUsersWithHandle)
+ <> map (mel,) (fst <$> Map.elems usersInPhase.personalUsersWithoutHandle)
+
+ checkAllDeletionsWorked :: (HasCallStack) => String -> Value -> IntMap TestUserList -> Int -> App ()
+ checkAllDeletionsWorked domain mel seedUsers phase = do
+ let deletedSoFar = IntMap.restrictKeys seedUsers (IntSet.fromList $ [1 .. phase])
+ pooledForConcurrentlyN_ parallelism (scimUserIdsWithGetter deletedSoFar) $ \(token, uid) ->
+ getScimUser domain token uid >>= assertStatus 404
+
+ pooledForConcurrentlyN_ parallelism (nonScimUserIds mel deletedSoFar) $ \(getter, user) -> do
+ uid <- user %. "qualified_id.id" & asString
+ mTid <- lookupField user "team" & asStringM
+ getUser getter (object ["domain" .= domain, "id" .= uid]) `bindResponse` \resp -> do
+ resp.status `shouldMatchInt` 200
+ resp.json `shouldMatch` tombstone domain uid mTid
+
+ let bots =
+ concatMap
+ ( \testUsers ->
+ map fst (Map.elems testUsers.botsInTeamConvs.users)
+ <> map fst (Map.elems testUsers.botsInPersonalConvs)
+ )
+ (IntMap.elems deletedSoFar)
+ pooledForConcurrentlyN_ parallelism bots $ \botUser -> do
+ botTombstone <- setField "status" "deleted" =<< setField "deleted" True botUser
+ getSelf botUser `bindResponse` \resp -> do
+ resp.status `shouldMatchInt` 200
+ resp.json `shouldMatch` botTombstone
+
+ checkUnaffectedUsers :: (HasCallStack) => String -> IntMap TestUserList -> IntMap TestUserList -> Int -> App ()
+ checkUnaffectedUsers domain seedUsersToBeUpdated seedUsersToBeDeleted phase = do
+ let upcomingPhases = IntSet.fromList [(phase + 1) .. 5]
+ usersNotYetDeleted = IntMap.restrictKeys seedUsersToBeDeleted upcomingPhases
+ usersNotYetUpdated = IntMap.restrictKeys seedUsersToBeUpdated upcomingPhases
+ existingUserLists = IntMap.elems usersNotYetDeleted <> IntMap.elems usersNotYetUpdated
+ existingScimUsers = concatMap extractScimUsers existingUserLists
+ existingUsers = concatMap extractTestUsers existingUserLists
+
+ checkScimUsers domain existingScimUsers
+ checkUsers existingUsers
+
+ checkNewAndUpdatedUsers :: (HasCallStack) => String -> IntMap TestUserList -> IntMap TestUserList -> App ()
+ checkNewAndUpdatedUsers domain newUsers updatedUsers = do
+ let scimUsers =
+ concatMap extractScimUsers newUsers
+ <> concatMap extractScimUsers updatedUsers
+ users =
+ concatMap extractTestUsers newUsers
+ <> concatMap extractTestUsers updatedUsers
+ checkScimUsers domain scimUsers
+ checkUsers users
+
+ checkScimUsers :: (HasCallStack) => String -> [(String, Value)] -> App ()
+ checkScimUsers domain tokensAndUsers = do
+ pooledForConcurrentlyN_ parallelism tokensAndUsers $ \(token, scimUser) ->
+ addJSONToFailureContext "scimUser" scimUser $ do
+ uid <- scimUser %. "id" & asString
+ getScimUser domain token uid `bindResponse` \resp -> do
+ resp.status `shouldMatchInt` 200
+ resp.json `shouldMatch` scimUser
+
+ checkUsers :: (HasCallStack) => [Value] -> App ()
+ checkUsers users =
+ pooledForConcurrentlyN_ parallelism users $ \user ->
+ addJSONToFailureContext "user" user $ do
+ getSelf user `bindResponse` \resp -> do
+ resp.status `shouldMatchInt` 200
+ resp.json `shouldMatch` user
+
+ extractScimUsers :: TestUserList -> [(String, Value)]
+ extractScimUsers testUserList =
+ map (testUserList.scimUsersWithRichInfo.token,) (fst3 <$> Map.elems testUserList.scimUsersWithRichInfo.users)
+ <> map (testUserList.scimUsersWithoutRichInfo.token,) (fst3 <$> Map.elems testUserList.scimUsersWithoutRichInfo.users)
+ <> map (testUserList.pendingScimUsers.token,) (fst3 <$> Map.elems testUserList.pendingScimUsers.users)
+
+ extractTestUsers :: TestUserList -> [Value]
+ extractTestUsers testUserList =
+ map thd3 (Map.elems testUserList.scimUsersWithRichInfo.users)
+ <> map thd3 (Map.elems testUserList.scimUsersWithoutRichInfo.users)
+ <> map fst (Map.elems testUserList.ssoUsers.users)
+ <> map fst (Map.elems testUserList.passwordTeamUsers.users)
+ <> map fst (Map.elems testUserList.personalUsersWithHandle)
+ <> map fst (Map.elems testUserList.personalUsersWithoutHandle)
+ <> extractBots testUserList
+
+ extractBots :: TestUserList -> [Value]
+ extractBots testUserList =
+ map fst (Map.elems testUserList.botsInTeamConvs.users)
+ <> map fst (Map.elems testUserList.botsInPersonalConvs)
+
+ createTestUsers :: (HasCallStack, MakesValue mel) => String -> mel -> String -> String -> Int -> App TestUserList
+ createTestUsers domain mel pid sid n = runConcurrently $ do
+ scimUsersWithRichInfo <- Concurrently $ createScimUsers domain n True True
+ scimUsersWithoutRichInfo <- Concurrently $ createScimUsers domain n False True
+ pendingScimUsers <- Concurrently $ createScimUsers domain n False False
+ ssoUsers <- Concurrently $ createSsoUsers domain n
+ passwordTeamUsers <- Concurrently $ createPasswordTeamUsers domain n
+ personalUsersWithoutHandle <- Concurrently $ createPersonalUsers domain mel n False
+ personalUsersWithHandle <- Concurrently $ createPersonalUsers domain mel n True
+ botsInTeamConvs <- Concurrently $ createTeamBots domain pid sid n
+ botsInPersonalConvs <- Concurrently $ createConvsAndAddBot domain mel Nothing pid sid n
+ pure TestUserList {..}
+
+ getUnqualifiedUser :: String -> String -> App (Map String Value)
+ getUnqualifiedUser domain uid = do
+ let quid = object ["domain" .= domain, "id" .= uid]
+ Map.singleton uid <$> (getSelf quid >>= getJSON 200)
+
+ createScimUsers :: (HasCallStack) => String -> Int -> Bool -> Bool -> App TestScimUsers
+ createScimUsers domain n shouldCreateRichInfo shouldAcceptInvite = do
+ (owner, tid, _) <- createTeam domain 1
+ tok <- createScimToken owner def >>= \resp -> resp.json %. "token" >>= asString
+ users <- fmap Map.unions . pooledReplicateConcurrentlyN 16 n $ do
+ newScimUser0 <- randomScimUser
+ newScimUser <-
+ if shouldCreateRichInfo
+ then do
+ richInfoKey <- randomAlphaString 10
+ richInfoValue <- randomString 10
+ modifyObject (KeyMap.insert (fromString "urn:ietf:params:scim:schemas:extension:wire:1.0:User") (object [richInfoKey .= richInfoValue]))
+ =<< setField "schemas" ["urn:ietf:params:scim:schemas:core:2.0:User", "urn:ietf:params:scim:schemas:extension:wire:1.0:User"] newScimUser0
+ else pure newScimUser0
+ email <- asString $ newScimUser %. "emails.0.value"
+ inactiveScimUser <- createScimUser domain tok newScimUser >>= getJSON 201
+ uid <- inactiveScimUser %. "id" & asString
+ (scimUser, userOrInv) <-
+ if shouldAcceptInvite
+ then do
+ registerInvitedUser domain tid email
+ scimUser <- getScimUser owner tok uid >>= getJSON 200
+ (scimUser,) <$> getUnqualifiedUser domain uid
+ else fmap (inactiveScimUser,) . fmap (Map.singleton uid) . getJSON 200 =<< I.getInvitationByEmail domain email
+ pure $ (scimUser,defPassword,) <$> userOrInv
+ pure $ TestScimUsers owner tok users
+
+ deleteScimUsers :: (HasCallStack) => String -> Bool -> TestScimUsers -> App ()
+ deleteScimUsers domain arePendingUsers testScimUsers = do
+ withWebSocket testScimUsers.owner $ \wsOwner -> do
+ pooledForConcurrentlyN_ parallelism testScimUsers.users $ \(scimUser, _, _) -> do
+ uid <- scimUser %. "id" & asString
+ deleteScimUser domain testScimUsers.token uid >>= assertSuccess
+ getScimUser domain testScimUsers.token uid >>= assertStatus 404
+
+ unless arePendingUsers $ do
+ void $ awaitNMatches (Map.size testScimUsers.users) isTeamMemberLeaveNotif wsOwner
+
+ updatePendingScimUserAndCheck :: (HasCallStack) => String -> String -> (Value, String, Value) -> UserUpdate -> App (Value, String, Value)
+ updatePendingScimUserAndCheck domain token (scimUser, pw, inv) update = do
+ addJSONToFailureContext "update" update . addJSONToFailureContext "scimUser" scimUser $ do
+ uid <- scimUser %. "id" & asString
+ updatedScimUser <- case update of
+ UpdatePassword _ -> do
+ pure scimUser
+ _ -> do
+ let updateScimRecord = case update of
+ UpdateName newName -> setField "displayName" newName
+ UpdateEmail newEmail -> setField "emails" (Array (Vector.singleton (object ["value" .= newEmail])))
+ UpdateHandle newHandle -> setField "userName" newHandle
+ updateScimReq <- setField "active" True =<< updateScimRecord scimUser
+ updateScimUser domain token uid updateScimReq `bindResponse` \resp -> do
+ resp.status `shouldMatchInt` 200
+ case update of
+ UpdateName newName -> resp.json %. "displayName" `shouldMatch` newName
+ UpdateEmail newEmail -> resp.json %. "emails.0.value" `shouldMatch` newEmail
+ UpdateHandle newHandle -> resp.json %. "userName" `shouldMatch` newHandle
+ assertJust "expected a updated scim user" resp.json
+ (updatedUserOrInv, newPassword) <- do
+ case update of
+ UpdatePassword newPassword -> pure (inv, newPassword)
+ _ ->
+ -- Changing email of a pending user doesn't generate a new
+ -- invitation, perhaps this is a bug?
+ -- Changing other things ofc doesn't generate a new invitation.
+ pure (inv, pw)
+ pure (updatedScimUser, newPassword, updatedUserOrInv)
+
+ updateScimUserAndCheck :: (HasCallStack) => String -> String -> (Value, String, Value) -> UserUpdate -> App (Value, String, Value)
+ updateScimUserAndCheck domain token (scimUser, pw, user) update = do
+ uid <- scimUser %. "id" & asString
+ updatedScimUser <- case update of
+ UpdatePassword newPassword -> do
+ putPassword user pw newPassword >>= assertSuccess
+ pure scimUser
+ _ -> do
+ updateScimReq <- case update of
+ UpdateName newName -> setField "displayName" newName scimUser
+ UpdateEmail newEmail -> setField "emails" (Array (Vector.singleton (object ["value" .= newEmail]))) scimUser
+ UpdateHandle newHandle -> setField "userName" newHandle scimUser
+ updateScimUser domain token uid updateScimReq `bindResponse` \resp -> do
+ resp.status `shouldMatchInt` 200
+ case update of
+ UpdateName newName -> resp.json %. "displayName" `shouldMatch` newName
+ UpdateEmail newEmail -> resp.json %. "emails.0.value" `shouldMatch` newEmail
+ UpdateHandle newHandle -> resp.json %. "userName" `shouldMatch` newHandle
+ assertJust "expected a updated scim user" resp.json
+ (updatedUserOrInv, newPassword) <- case update of
+ UpdatePassword newPassword -> do
+ email <- scimUser %. "emails.0.value" & asString
+ login domain email newPassword >>= assertSuccess
+ pure (user, newPassword)
+ UpdateEmail newEmail -> do
+ activateEmail domain newEmail
+ getSelf user `bindResponse` \resp -> do
+ resp.status `shouldMatchInt` 200
+ resp.json %. "email" `shouldMatch` newEmail
+ (,pw) <$> assertJust "expected user data" resp.json
+ _ -> do
+ getSelf user `bindResponse` \resp -> do
+ resp.status `shouldMatchInt` 200
+ case update of
+ UpdateName newName -> resp.json %. "name" `shouldMatch` newName
+ UpdateHandle newHandle -> resp.json %. "handle" `shouldMatch` newHandle
+ (,pw) <$> assertJust "expected user data" resp.json
+ pure (updatedScimUser, newPassword, updatedUserOrInv)
+
+ updateScimUsers :: (HasCallStack) => String -> TestScimUsers -> [UserUpdate] -> App TestScimUsers
+ updateScimUsers domain testScimUsers updates = do
+ let usersWithUpdates = (zip (Map.elems testScimUsers.users) updates)
+ updatedUsers <- fmap Map.unions . pooledForConcurrentlyN parallelism usersWithUpdates $ \((scimUser, pw, user), update) -> do
+ uid <- scimUser %. "id" & asString
+ Map.singleton uid <$> updateScimUserAndCheck domain testScimUsers.token (scimUser, pw, user) update
+
+ pure $ (testScimUsers {users = updatedUsers} :: TestScimUsers)
+
+ updatePendingScimUsers :: (HasCallStack) => String -> TestScimUsers -> [PendingScimUpdate] -> App TestScimUsers
+ updatePendingScimUsers domain testScimUsers updates = do
+ let usersWithUpdates = (zip (Map.elems testScimUsers.users) updates)
+ updatedUsers <- fmap Map.unions . pooledForConcurrentlyN parallelism usersWithUpdates $ \((scimUser, pw, inv), update) -> do
+ uid <- scimUser %. "id" & asString
+ email <- scimUser %. "externalId" & asString
+ tid <- testScimUsers.owner %. "team" & asString
+ Map.singleton uid <$> case update of
+ RegisterPendingScimUser -> do
+ registerInvitedUser domain tid email
+ updatedScimUser <- getScimUser domain testScimUsers.token uid >>= getJSON 200
+ let quid = object ["domain" .= domain, "id" .= uid]
+ fmap (updatedScimUser,pw,) . getJSON 200 =<< getSelf quid
+ UpdatePendingScimUser updateUser -> do
+ updatePendingScimUserAndCheck domain testScimUsers.token (scimUser, pw, inv) updateUser
+ pure (testScimUsers {users = updatedUsers} :: TestScimUsers)
+
+ createSsoUsers :: (HasCallStack) => String -> Int -> App TestTeamUsers
+ createSsoUsers domain n = do
+ (owner, tid, _) <- createTeam domain 1
+ I.setTeamFeatureStatus owner tid "sso" "enabled" >>= assertSuccess
+ (createIdpResp, (idpMeta, privcreds)) <- registerTestIdPWithMetaWithPrivateCreds owner
+ assertSuccess createIdpResp
+ idpId <- asString =<< (createIdpResp.json %. "id")
+
+ users <- fmap Map.unions . pooledReplicateConcurrentlyN 16 n $ do
+ subject <- nextSubject
+ (mUid, _) <- loginWithSamlWithZHost Nothing domain True tid subject (idpId, (idpMeta, privcreds))
+ uid <- assertJust "user id not created by logging in with SAML" mUid
+ (,Nothing) <$$> getUnqualifiedUser domain uid
+ pure $ TestTeamUsers {..}
+
+ createPasswordTeamUsers :: (HasCallStack) => String -> Int -> App TestTeamUsers
+ createPasswordTeamUsers domain n = do
+ (owner, _tid, usersWithoutPassword) <- createTeam domain n
+
+ users <- fmap Map.unions . pooledForConcurrentlyN parallelism usersWithoutPassword $ \user -> do
+ p <- randomPassword
+ putPassword user defPassword p >>= assertSuccess
+ uid <- user %. "qualified_id.id" & asString
+ pure $ Map.singleton uid (user, Just p)
+
+ pure $ TestTeamUsers {..}
+
+ deleteTeamUsers :: (HasCallStack) => TestTeamUsers -> App ()
+ deleteTeamUsers team = do
+ withWebSocket team.owner $ \wsOwner -> do
+ tid <- team.owner %. "team" & asString
+ pooledForConcurrentlyN_ parallelism team.users $ \(user, _) -> do
+ uid <- user %. "qualified_id.id" & asString
+ deleteTeamMember tid team.owner uid >>= assertSuccess
+
+ void $ awaitNMatches (Map.size team.users) isTeamMemberLeaveNotif wsOwner
+
+ getSelfWithAssertion :: (HasCallStack, MakesValue user) => user -> ((HasCallStack) => Response -> App ()) -> App (Map String Value)
+ getSelfWithAssertion user assertion = do
+ getSelf user `bindResponse` \resp -> do
+ resp.status `shouldMatchInt` 200
+ assertion resp
+ Map.singleton <$> (resp.json %. "qualified_id.id" & asString) <*> (assertJust "expected GET /self to return a JSON" resp.json)
+
+ checkUpdateUser :: (HasCallStack) => Map String (Value, Maybe String) -> [UserUpdate] -> App (Map String (Value, Maybe String))
+ checkUpdateUser users updates = do
+ fmap Map.unions . pooledForConcurrentlyN parallelism (zip (Map.elems users) updates) $ \((user, mPassword), update) ->
+ addJSONToFailureContext "user" user . addJSONToFailureContext "update" update $ do
+ updatedUser <- case (update, mPassword) of
+ (UpdateName newName, _) -> do
+ putSelf user def {name = Just newName} >>= assertSuccess
+ getSelfWithAssertion user $ \resp -> resp.json %. "name" `shouldMatch` newName
+ (UpdateEmail newEmail, Just pw) -> do
+ oldEmail <- user %. "email" & asString
+ (cookie, token) <- bindResponse (login user oldEmail pw) $ \resp -> do
+ resp.status `shouldMatchInt` 200
+ token <- resp.json %. "access_token" & asString
+ let cookie = fromJust $ getCookie "zuid" resp
+ pure ("zuid=" <> cookie, token)
+ updateEmail user newEmail cookie token >>= assertSuccess
+ activateEmail user newEmail
+ getSelfWithAssertion user $ \resp -> resp.json %. "email" `shouldMatch` newEmail
+ (UpdateEmail {}, Nothing) -> do
+ uid <- user %. "qualified_id.id" & asString
+ pure $ Map.singleton uid user
+ (UpdateHandle newHandle, _) -> do
+ putHandle user newHandle >>= assertSuccess
+ getSelfWithAssertion user $ \resp -> resp.json %. "handle" `shouldMatch` newHandle
+ (UpdatePassword newPassword, Just oldPassword) -> do
+ email <- user %. "email" & asString
+ putPassword user oldPassword newPassword >>= assertSuccess
+ login user email oldPassword `bindResponse` \resp ->
+ resp.status `shouldMatchInt` 403
+ login user email newPassword >>= assertSuccess
+ uid <- user %. "qualified_id.id" & asString
+ pure $ Map.singleton uid user
+ (UpdatePassword {}, Nothing) -> do
+ uid <- user %. "qualified_id.id" & asString
+ pure $ Map.singleton uid user
+ pure $ (,mPassword) <$> updatedUser
+
+ createPersonalUsers :: (HasCallStack, MakesValue mel) => String -> mel -> Int -> Bool -> App (Map String (Value, Maybe String))
+ createPersonalUsers domain mel n claimHandle =
+ fmap Map.unions . pooledReplicateConcurrentlyN parallelism n $ do
+ user <- randomUser domain def
+ connectTwoUsers mel user
+ uid <- user %. "qualified_id.id" & asString
+ if claimHandle
+ then do
+ hdl <- randomHandle
+ putHandle user hdl >>= assertSuccess
+ fmap (,Just defPassword) . Map.singleton uid <$> (setField "handle" hdl user)
+ else pure $ Map.singleton uid (user, Just defPassword)
+
+ deletePersonalUsers :: (HasCallStack, MakesValue mel, ToWSConnect mel) => mel -> Map String (Value, Maybe String) -> App ()
+ deletePersonalUsers mel users =
+ withWebSocket mel $ \wsMel -> do
+ pooledForConcurrentlyN_ parallelism users $ uncurry deleteUserWithPassword
+ void $ awaitNMatches (Map.size users) isDeleteUserNotif wsMel
+
+ createConvsAndAddBot :: (HasCallStack, MakesValue user) => String -> user -> Maybe String -> String -> String -> Int -> App (Map String (Value, Value))
+ createConvsAndAddBot domain user tid pid sid n = do
+ fmap Map.unions . pooledReplicateConcurrentlyN parallelism n $ do
+ conv <- postConversation user (defProteus {team = tid}) >>= getJSON 201
+ convId <- conv %. "qualified_id" & objId
+ addBotResp <- addBot user pid sid convId >>= getJSON 201
+ botId <- addBotResp %. "id" & asString
+ (,conv) <$$> getUnqualifiedUser domain botId
+
+ createTeamBots :: (HasCallStack) => String -> String -> String -> Int -> App TestTeamUsers
+ createTeamBots domain pid sid n = do
+ (owner, tid, _) <- createTeam domain 1
+ postServiceWhitelist owner tid (object ["id" .= sid, "provider" .= pid, "whitelisted" .= True])
+ >>= assertSuccess
+ TestTeamUsers owner . fmap (\(x, _) -> (x, Nothing)) <$> createConvsAndAddBot domain owner (Just tid) pid sid n
+
+ deleteBotsTeam :: (HasCallStack) => TestTeamUsers -> String -> String -> App ()
+ deleteBotsTeam testTeam pid sid = do
+ tid <- testTeam.owner %. "team" & asString
+ withWebSocket testTeam.owner $ \ws -> do
+ postServiceWhitelist testTeam.owner tid (object ["id" .= sid, "provider" .= pid, "whitelisted" .= False]) >>= assertSuccess
+ void $ awaitNMatches (Map.size testTeam.users) isConvLeaveNotif ws
+
+ deleteBotConvs :: (HasCallStack) => Value -> Map String (Value, Value) -> App ()
+ deleteBotConvs mel botConvs = do
+ pooledForConcurrentlyN_ parallelism (Map.elems botConvs) $ \(bot, conv) -> do
+ cid <- conv %. "qualified_id.id" & asString
+ bid <- bot %. "qualified_id.id" & asString
+ rmBotSelf mel bid cid >>= assertSuccess
+
+-- | This test creates users in PG and Cassandra separately to simulate a
+-- situation where there are users in both DBs. Then tries to index them into ES
+-- to make sure the pagination over these users works.
+testReindexingUsersDuringMigration :: (HasCallStack) => App ()
+testReindexingUsersDuringMigration = do
+ resourcePool <- asks (.resourcePool)
+
+ runCodensity (acquireResources 1 resourcePool) $ \[backend] -> do
+ let domain = backend.berDomain
+ -- Create users in cassandra using 'phase1Overrides'
+ (casSearcher, casExistingUsers, casDeletedUsers) <-
+ runCodensity (startDynamicBackend backend phase1Overrides)
+ $ \_ -> setupUsers domain
+
+ -- Create users in postgres using 'phase5Overrides'
+ (pgSearcher, pgExistingUsers, pgDeletedUsers) <-
+ runCodensity (startDynamicBackend backend phase5Overrides)
+ $ \_ -> setupUsers domain
+
+ -- Test that searching in the already existing index works with in
+ -- 'phase2Overrides', which should work with data in cassandra and postgres
+ runCodensity (startDynamicBackend backend phase2Overrides) $ \_ -> do
+ I.refreshIndex domain
+ checkSearchWorks domain casSearcher casExistingUsers casDeletedUsers
+ checkSearchWorks domain pgSearcher pgExistingUsers pgDeletedUsers
+
+ newIndex <- createNewIndex
+ let backendWithNewIndex = backend {berElasticsearchIndex = newIndex}
+ runCodensity (startDynamicBackend backendWithNewIndex phase2Overrides) $ \_ -> do
+ reindexUsers backendWithNewIndex phase2Overrides 5
+ I.refreshIndex domain
+ checkSearchWorks domain casSearcher casExistingUsers casDeletedUsers
+ checkSearchWorks domain pgSearcher pgExistingUsers pgDeletedUsers
+ where
+ n = 5
+ parallelism = 16
+
+ setupUsers :: (HasCallStack) => String -> App (Value, [Value], [Value])
+ setupUsers domain = do
+ searcher <- randomUser domain def
+ existingUsers <- pooledReplicateConcurrentlyN parallelism n $ randomUser domain def
+ deletedUsers <- pooledReplicateConcurrentlyN parallelism n $ do
+ u <- randomUser domain def
+ connectTwoUsers searcher u
+ pure u
+ withWebSocket searcher $ \ws -> do
+ pooledForConcurrentlyN_ parallelism deletedUsers deleteUser
+ void $ awaitNMatches n isDeleteUserNotif ws
+ pure (searcher, existingUsers, deletedUsers)
+
+ checkSearchWorks :: (HasCallStack) => String -> Value -> [Value] -> [Value] -> App ()
+ checkSearchWorks domain searcher existingUsers deletedUsers = do
+ pooledForConcurrentlyN_ parallelism existingUsers $ \u ->
+ assertCanFind searcher u (u %. "name") domain
+
+ pooledForConcurrentlyN_ parallelism deletedUsers $ \u ->
+ assertCannotFind searcher u (u %. "name") domain
+
+-- handleA: Alice and Anna have the same handle, but the handle claims table
+-- supports Alice's claim. After the migration Bob loses their handle.
+--
+-- handleB: Bob and Bill also have the same handle, but the handle claims table
+-- doesn't support any of their claims. After the migration both of them will
+-- loose the claim.
+--
+-- handleC: Carl and Creed also have the same handle, Cassandra supports Carl's
+-- claim, while Postgresql supports Creed's claim. In this case Creed gets to
+-- keep their handle.
+testMigrationOfUsersWithHandleDisputes :: (HasCallStack) => App ()
+testMigrationOfUsersWithHandleDisputes = do
+ resourcePool <- asks (.resourcePool)
+ -- Between Alice and Anna
+ handleA <- randomHandle
+
+ -- In user record for Bob and Bill
+ handleB <- randomHandle
+
+ -- Between Carl and Creed
+ handleC <- randomHandle
+
+ runCodensity (acquireResources 1 resourcePool) $ \[backend] -> do
+ let domain = backend.berDomain
+ brigKeyspace = backend.berBrigKeyspace
+ (alice, anna, bob, bill, carl) <- runCodensity (startDynamicBackend backend phase1Overrides) $ \_ -> do
+ alice <- randomUser domain def
+ anna <- randomUser domain def
+ bob <- randomUser domain def
+ bill <- randomUser domain def
+ carl <- randomUser domain def
+
+ Just annaId <- UUID.fromString <$> (anna %. "qualified_id.id" & asString)
+ Just bobId <- UUID.fromString <$> (bob %. "qualified_id.id" & asString)
+ Just billId <- UUID.fromString <$> (bill %. "qualified_id.id" & asString)
+
+ -- Claim handle correctly for alice
+ putHandle alice handleA >>= assertSuccess
+ putHandle carl handleC >>= assertSuccess
+
+ -- Claim handle by hacking into the DB for others. There seems to be no
+ -- other way of testing this edge case
+ let assignHandleQuery :: PrepQuery W (Text, UUID) () = fromString $ "UPDATE " <> brigKeyspace <> ".user SET handle = ? WHERE id = ?"
+ write assignHandleQuery $ defQueryParams LocalQuorum (Text.pack handleA, annaId)
+ write assignHandleQuery $ defQueryParams LocalQuorum (Text.pack handleB, bobId)
+ write assignHandleQuery $ defQueryParams LocalQuorum (Text.pack handleB, billId)
+
+ assertHandle alice (Just handleA)
+ assertHandle anna (Just handleA)
+ assertHandle bob (Just handleB)
+ assertHandle bill (Just handleB)
+ assertHandle carl (Just handleC)
+
+ pure (alice, anna, bob, bill, carl)
+
+ -- Start Phase 5 here so that we can claim the same handle for Dan as Doug
+ -- but in Postgresql. The production scenario can only happen due to a race
+ -- condition. This is just a more precise way of causing the DB
+ -- inconsistency.
+ creed <- runCodensity (startDynamicBackend backend phase5Overrides) $ \_ -> do
+ creed <- randomUser domain def
+ putHandle creed handleC >>= assertSuccess
+ assertHandle creed (Just handleC)
+ pure creed
+
+ runCodensity (startDynamicBackend backend phase3Overrides) $ \_ -> do
+ waitForMigration domain userMigrationFinishedCounterName
+ assertMigrationSuccessful domain "^wire_users_migration_failed"
+
+ runCodensity (startDynamicBackend backend phase5Overrides) $ \_ -> do
+ assertHandle alice (Just handleA)
+ assertHandle anna Nothing
+
+ assertHandle bob Nothing
+ assertHandle bill Nothing
+
+ assertHandle carl Nothing
+ assertHandle creed (Just handleC)
+
+ -- handleA cannot be claimed
+ putHandle anna handleA >>= assertStatus 409
+
+ -- handleB can be claimed
+ putHandle bob handleB >>= assertSuccess
+
+ -- handleC cannot be claimed
+ putHandle carl handleC >>= assertStatus 409
+ where
+ assertHandle :: (HasCallStack) => Value -> Maybe String -> App ()
+ assertHandle user expectedHandle = do
+ getSelf user `bindResponse` \resp -> do
+ resp.status `shouldMatchInt` 200
+ case expectedHandle of
+ Just h ->
+ resp.json %. "handle" `shouldMatch` h
+ Nothing ->
+ case resp.json of
+ Just (Object o) -> KM.keys o `shouldNotContain` [fromString "handle"]
+ _ -> assertFailure "Unexpected body for getSelf"
+
+testMigrationOfInvalidUsers :: (HasCallStack) => App ()
+testMigrationOfInvalidUsers = do
+ resourcePool <- asks (.resourcePool)
+
+ runCodensity (acquireResources 1 resourcePool) $ \[backend] -> do
+ let domain = backend.berDomain
+ brigKeyspace = backend.berBrigKeyspace
+ (validUser, noName, noNameId, noActivated, noActivatedId) <- runCodensity (startDynamicBackend backend phase1Overrides) $ \_ -> do
+ validUser <- randomUser domain def
+
+ noName <- randomUser domain def
+ Just noNameId <- UUID.fromString <$> (noName %. "qualified_id.id" & asString)
+
+ noActivated <- randomUser domain def
+ Just noActivatedId <- UUID.fromString <$> (noActivated %. "qualified_id.id" & asString)
+
+ -- Cause users to be invalid by poking into Cassandra
+ let removeName :: PrepQuery W (Identity UUID) () = fromString $ "UPDATE " <> brigKeyspace <> ".user SET name = NULL WHERE id = ?"
+ removeActivated :: PrepQuery W (Identity UUID) () = fromString $ "UPDATE " <> brigKeyspace <> ".user SET activated = NULL WHERE id = ?"
+ write removeName $ defQueryParams LocalQuorum (Identity noNameId)
+ write removeActivated $ defQueryParams LocalQuorum (Identity noActivatedId)
+
+ getSelf validUser >>= assertStatus 200
+ getSelf noName >>= assertStatus 500
+ getSelf noActivated >>= assertStatus 500
+
+ pure (validUser, noName, noNameId, noActivated, noActivatedId)
+
+ runCodensity (startDynamicBackend backend phase3Overrides) $ \_ -> do
+ waitForMigration domain userMigrationFinishedCounterName
+
+ runCodensity (startDynamicBackend backend phase5Overrides) $ \_ -> do
+ getSelf validUser >>= assertStatus 200
+ getSelf noName >>= assertStatus 404
+ getSelf noActivated >>= assertStatus 404
+
+ -- Delete invalid users from cassandra so they don't trip other tests. These
+ -- other tests are usually reindexing the users, the reindex code doesn't
+ -- deal with invalid users so well.
+ let deleteUserRow :: PrepQuery W (Identity UUID) () = fromString $ "DELETE FROM " <> brigKeyspace <> ".user WHERE id = ?"
+ write deleteUserRow $ defQueryParams LocalQuorum (Identity noNameId)
+ write deleteUserRow $ defQueryParams LocalQuorum (Identity noActivatedId)
+
+-- * Test Helpers
+
+data TestUsersByOperations = TestUsersByOperations
+ { updates :: IntMap TestUserList,
+ deletes :: IntMap TestUserList
+ }
+ deriving (Show, Eq, Generic)
+
+instance Semigroup TestUsersByOperations where
+ users1 <> users2 =
+ TestUsersByOperations
+ { updates = users1.updates <> users2.updates,
+ deletes = users1.deletes <> users2.deletes
+ }
+
+instance Monoid TestUsersByOperations where
+ mempty = TestUsersByOperations {updates = mempty, deletes = mempty}
+
+instance ToJSON TestUsersByOperations
+
+data TestUserList = TestUserList
+ { scimUsersWithRichInfo :: TestScimUsers,
+ scimUsersWithoutRichInfo :: TestScimUsers,
+ pendingScimUsers :: TestScimUsers,
+ ssoUsers :: TestTeamUsers,
+ passwordTeamUsers :: TestTeamUsers,
+ personalUsersWithoutHandle :: Map String (Value, Maybe String),
+ personalUsersWithHandle :: Map String (Value, Maybe String),
+ botsInTeamConvs :: TestTeamUsers,
+ -- UserId -> (User, Conv)
+ botsInPersonalConvs :: Map String (Value, Value)
+ }
+ deriving (Show, Eq)
+
+data TestScimUsers = TestScimUsers
+ { owner :: Value,
+ token :: String,
+ -- | ScimUser, Password, UserOrInv
+ users :: Map String (Value, String, Value)
+ }
+ deriving (Show, Eq)
+
+data TestTeamUsers = TestTeamUsers
+ { owner :: Value,
+ -- | (user, maybe password)
+ users :: Map String (Value, Maybe String)
+ }
+ deriving (Show, Eq)
+
+instance ToJSON TestUserList where
+ toJSON userList = do
+ object
+ [ fromString "scimUsersWithRichInfo" .= Map.keys userList.scimUsersWithRichInfo.users,
+ fromString "scimUsersWithoutRichInfo" .= Map.keys userList.scimUsersWithoutRichInfo.users,
+ fromString "pendingScimUsers" .= Map.keys userList.pendingScimUsers.users,
+ fromString "ssoUsers" .= Map.keys userList.ssoUsers.users,
+ fromString "passwordTeamUsers" .= Map.keys userList.passwordTeamUsers.users,
+ fromString "personalUsersWithoutHandle" .= Map.keys userList.personalUsersWithoutHandle,
+ fromString "personalUsersWithHandle" .= Map.keys userList.personalUsersWithHandle,
+ fromString "botsInTeamConvs" .= Map.keys userList.botsInTeamConvs.users,
+ fromString "botsInPersonalConvs" .= Map.keys userList.botsInPersonalConvs
+ ]
+
+data UserUpdate
+ = UpdateName String
+ | UpdateHandle String
+ | UpdateEmail String
+ | UpdatePassword String
+ deriving (Show, Eq, Generic)
+
+instance Arbitrary UserUpdate where
+ arbitrary =
+ oneof
+ [ UpdateName <$> arbitraryName,
+ UpdateHandle <$> arbitraryHandle,
+ UpdateEmail <$> arbitraryEmail,
+ UpdatePassword <$> arbitraryPassword
+ ]
+
+instance ToJSON UserUpdate
+
+arbitraryNonPasswordUpdate :: Gen UserUpdate
+arbitraryNonPasswordUpdate =
+ oneof
+ [ UpdateName <$> arbitraryName,
+ UpdateHandle <$> arbitraryHandle,
+ UpdateEmail <$> arbitraryEmail
+ ]
+
+data PendingScimUpdate
+ = RegisterPendingScimUser
+ | UpdatePendingScimUser UserUpdate
+ deriving (Show, Eq, Generic)
+
+instance Arbitrary PendingScimUpdate where
+ arbitrary =
+ oneof
+ [ pure RegisterPendingScimUser,
+ UpdatePendingScimUser <$> arbitraryNonPasswordUpdate
+ ]
+
+instance ToJSON PendingScimUpdate
+
+data PhaseUpdates = PhaseUpdates
+ { scimUsersWithRichInfo :: [UserUpdate],
+ scimUsersWithoutRichInfo :: [UserUpdate],
+ pendingScimUsers :: [PendingScimUpdate],
+ ssoUsers :: [UserUpdate],
+ passwordTeamUsers :: [UserUpdate],
+ personalUsersWithoutHandle :: [UserUpdate],
+ personalUsersWithHandle :: [UserUpdate]
+ }
+ deriving (Show, Eq, Generic)
+
+instance ToJSON PhaseUpdates
+
+arbitraryPhaseUpdates :: Int -> Gen PhaseUpdates
+arbitraryPhaseUpdates n = do
+ scimUsersWithRichInfo <- replicateM n arbitrary
+ scimUsersWithoutRichInfo <- replicateM n arbitrary
+ pendingScimUsers <- replicateM n arbitrary
+ ssoUsers <- replicateM n arbitraryNonPasswordUpdate
+ passwordTeamUsers <- replicateM n arbitrary
+ personalUsersWithoutHandle <- replicateM n arbitrary
+ personalUsersWithHandle <- replicateM n arbitrary
+ pure PhaseUpdates {..}
+
+userMigrationFinishedCounterName :: String
+userMigrationFinishedCounterName = "^wire_users_migration_finished"
+
+commonOverrides, phase1Overrides, phase2Overrides, phase3Overrides, phase4Overrides, phase5Overrides :: ServiceOverrides
+commonOverrides =
+ def
+ { brigCfg =
+ setField @_ @Int "optSettings.setUserMaxConnections" 500
+ >=> setField @_ @Int "optSettings.setActivationTimeout" 3600
+ >=> setField @_ @Int "optSettings.setVerificationTimeout" 3600
+ >=> setField @_ @Int "optSettings.setTeamInvitationTimeout" 3600
+ >=> setField @_ @Int "optSettings.setUserCookieRenewAge" 1209600
+ >=> setField @_ @Int "postgresqlPool.size" 200
+ >=> removeField "optSettings.setSuspendInactiveUsers"
+ }
+phase1Overrides =
+ commonOverrides
+ <> def
+ { brigCfg = setField "postgresMigration.user" "cassandra",
+ galleyCfg = setField "postgresMigration.user" "cassandra",
+ backgroundWorkerCfg =
+ setField "postgresMigration.user" "cassandra"
+ >=> setField "migrateUsers" False
+ }
+phase2Overrides =
+ commonOverrides
+ <> def
+ { brigCfg = setField "postgresMigration.user" "migration-to-postgresql",
+ galleyCfg = setField "postgresMigration.user" "migration-to-postgresql",
+ backgroundWorkerCfg =
+ setField "postgresMigration.user" "migration-to-postgresql"
+ >=> setField "migrateUsers" False
+ }
+phase3Overrides =
+ commonOverrides
+ <> def
+ { brigCfg = setField "postgresMigration.user" "migration-to-postgresql",
+ galleyCfg = setField "postgresMigration.user" "migration-to-postgresql",
+ backgroundWorkerCfg =
+ setField "postgresMigration.user" "migration-to-postgresql"
+ >=> setField "migrateUsers" True
+ }
+phase4Overrides =
+ commonOverrides
+ <> def
+ { brigCfg = setField "postgresMigration.user" "migration-to-postgresql",
+ galleyCfg = setField "postgresMigration.user" "migration-to-postgresql",
+ backgroundWorkerCfg =
+ setField "postgresMigration.user" "migration-to-postgresql"
+ >=> setField "migrateUsers" False
+ }
+phase5Overrides =
+ commonOverrides
+ <> def
+ { brigCfg = setField "postgresMigration.user" "postgresql",
+ galleyCfg = setField "postgresMigration.user" "postgresql",
+ backgroundWorkerCfg =
+ setField "postgresMigration.user" "postgresql"
+ >=> setField "migrateUsers" False
+ }
+
+phaseOverrides :: IntMap ServiceOverrides
+phaseOverrides =
+ IntMap.fromList
+ [ (1, phase1Overrides),
+ (2, phase2Overrides),
+ (3, phase3Overrides),
+ (4, phase4Overrides),
+ (5, phase5Overrides)
+ ]
diff --git a/integration/test/Test/Migration/Util.hs b/integration/test/Test/Migration/Util.hs
index ba3a116b453..28d1d4712ac 100644
--- a/integration/test/Test/Migration/Util.hs
+++ b/integration/test/Test/Migration/Util.hs
@@ -27,14 +27,30 @@ import GHC.Stack
import SetupHelpers hiding (deleteUser)
import Testlib.Prelude
import Text.Regex.TDFA ((=~))
+import UnliftIO
waitForMigration :: (HasCallStack) => String -> String -> App ()
-waitForMigration domain name = do
- metrics <-
- getMetrics domain BackgroundWorker `bindResponse` \resp -> do
- resp.status `shouldMatchInt` 200
- pure $ Text.decodeUtf8 resp.body
- let (_, _, _, finishedMatches) :: (Text, Text, Text, [Text]) = (metrics =~ Text.pack (name <> "\\ ([0-9]+\\.[0-9]+)$"))
- when (finishedMatches /= [Text.pack "1.0"]) $ do
- liftIO $ threadDelay 100_000
- waitForMigration domain name
+waitForMigration domain metricName =
+ maybe failWithContext pure =<< timeout 30_000_000 go
+ where
+ failWithContext = do
+ getMetrics domain BackgroundWorker `bindResponse` \resp -> do
+ resp.status `shouldMatchInt` 200
+ assertFailure "Timed out waiting for postgresql migration"
+ go = do
+ metrics <-
+ getMetrics domain BackgroundWorker `bindResponse` \resp -> do
+ resp.status `shouldMatchInt` 200
+ pure $ Text.decodeUtf8 resp.body
+ let (_, _, _, finishedMatches) :: (Text, Text, Text, [Text]) = (metrics =~ Text.pack (metricName <> "\\ ([0-9]+\\.[0-9]+)$"))
+ when (finishedMatches /= [Text.pack "1.0"]) $ do
+ liftIO $ threadDelay 100_000
+ go
+
+assertMigrationSuccessful :: (HasCallStack) => String -> String -> App ()
+assertMigrationSuccessful domain failedMetricName = do
+ getMetrics domain BackgroundWorker `bindResponse` \resp -> do
+ resp.status `shouldMatchInt` 200
+ let metrics = Text.decodeUtf8 resp.body
+ (_, _, _, failedMatches) :: (Text, Text, Text, [Text]) = (metrics =~ Text.pack (failedMetricName <> "\\ ([0-9]+\\.[0-9]+)$"))
+ failedMatches `shouldMatch` [Text.pack "0.0"]
diff --git a/integration/test/Test/Search.hs b/integration/test/Test/Search.hs
index a0986828555..b4b3edb4b79 100644
--- a/integration/test/Test/Search.hs
+++ b/integration/test/Test/Search.hs
@@ -693,7 +693,7 @@ testReindexAllUsers = do
assertCannotFind alice user (user %. "name") domain
-- Reindex users using a small page size so pagination gets excersiced
- reindexUsers testBackend 5
+ reindexUsers testBackend def 5
BrigI.refreshIndex domain
-- Now things should work as expected
diff --git a/integration/test/Testlib/Env.hs b/integration/test/Testlib/Env.hs
index dd37e9c8f56..6c0a774953c 100644
--- a/integration/test/Testlib/Env.hs
+++ b/integration/test/Testlib/Env.hs
@@ -96,14 +96,14 @@ mkGlobalEnv cfgFile = do
& Cassandra.setContacts intConfig.cassandra.cassHost []
& Cassandra.setPortNumber (fromIntegral intConfig.cassandra.cassPort)
cassSettings = maybe basicCassSettings (\sslCtx -> Cassandra.setSSLContext sslCtx basicCassSettings) mbSSLContext
- cassClient <- Cassandra.init cassSettings
+ gCassClient <- Cassandra.init cassSettings
let resources = backendResources (Map.elems intConfig.dynamicBackends)
resourcePool <-
liftIO $
createBackendResourcePool
resources
intConfig.rabbitmq
- cassClient
+ gCassClient
let sm =
Map.fromList $
[ (intConfig.backendOne.originDomain, intConfig.backendOne.beServiceMap),
@@ -146,7 +146,8 @@ mkGlobalEnv cfgFile = do
gDNSMockServerConfig = intConfig.dnsMockServer,
gCellsEventQueue = intConfig.cellsEventQueue,
gCellsEventWatchersLock,
- gCellsEventWatchers
+ gCellsEventWatchers,
+ gCassClient
}
where
createSSLContext :: Maybe FilePath -> IO (Maybe OpenSSL.SSLContext)
@@ -202,6 +203,7 @@ mkEnv currentTestName ge = do
cellsEventQueue = ge.gCellsEventQueue,
cellsEventWatchersLock = ge.gCellsEventWatchersLock,
cellsEventWatchers = ge.gCellsEventWatchers,
+ cassClient = ge.gCassClient,
curlTrace
}
diff --git a/integration/test/Testlib/JSON.hs b/integration/test/Testlib/JSON.hs
index 04ae74ef192..b02402cc27a 100644
--- a/integration/test/Testlib/JSON.hs
+++ b/integration/test/Testlib/JSON.hs
@@ -26,6 +26,7 @@ import Data.Aeson hiding ((.=))
import qualified Data.Aeson as Aeson
import qualified Data.Aeson.Encode.Pretty as Aeson
import qualified Data.Aeson.Key as KM
+import Data.Aeson.KeyMap (KeyMap)
import qualified Data.Aeson.KeyMap as KM
import qualified Data.Aeson.Types as Aeson
import Data.ByteString (ByteString)
@@ -320,6 +321,11 @@ modifyField selector up x = do
ob <- asObject v
pure $ Object $ KM.insert (KM.fromString k) newValue ob
+modifyObject :: (HasCallStack, MakesValue a) => (KeyMap Value -> KeyMap Value) -> a -> App Value
+modifyObject f x = do
+ ob <- asObject x
+ pure . Object $ f ob
+
-- | `removeField "a.b" {"a": {"b": 3}, "c": true} == {"a": {}, "c": true}`
removeField :: (HasCallStack, MakesValue a) => String -> a -> App Value
removeField selector x = do
diff --git a/integration/test/Testlib/Types.hs b/integration/test/Testlib/Types.hs
index 638c6a65279..2919803d459 100644
--- a/integration/test/Testlib/Types.hs
+++ b/integration/test/Testlib/Types.hs
@@ -55,6 +55,7 @@ import qualified Data.Text as T
import qualified Data.Text.Encoding as T
import Data.Time
import Data.Word
+import qualified Database.CQL.IO as Cassandra
import GHC.Generics (Generic)
import GHC.Records
import GHC.Stack
@@ -147,7 +148,8 @@ data GlobalEnv = GlobalEnv
gDNSMockServerConfig :: DNSMockServerConfig,
gCellsEventQueue :: String,
gCellsEventWatchersLock :: MVar (),
- gCellsEventWatchers :: IORef (Map String QueueWatcher)
+ gCellsEventWatchers :: IORef (Map String QueueWatcher),
+ gCassClient :: Cassandra.ClientState
}
data IntegrationConfig = IntegrationConfig
@@ -276,7 +278,8 @@ data Env = Env
cellsEventQueue :: String,
cellsEventWatchersLock :: MVar (),
cellsEventWatchers :: IORef (Map String QueueWatcher),
- curlTrace :: IORef [String]
+ curlTrace :: IORef [String],
+ cassClient :: Cassandra.ClientState
}
data Response = Response
@@ -488,6 +491,18 @@ newtype App a = App {unApp :: ReaderT Env IO a}
instance MonadRandom App where
getRandomBytes n = liftIO (getRandomBytes n)
+instance Cassandra.MonadClient App where
+ liftClient :: Cassandra.Client a -> App a
+ liftClient action = do
+ clientState <- asks (.cassClient)
+ liftIO $ Cassandra.runClient clientState action
+
+ localState :: (Cassandra.ClientState -> Cassandra.ClientState) -> App a -> App a
+ localState f action = do
+ env <- ask
+ let newClientState = f env.cassClient
+ liftIO $ runAppWithEnv (env {cassClient = newClientState}) action
+
runAppWithEnv :: Env -> App a -> IO a
runAppWithEnv e m = runReaderT (unApp m) e
diff --git a/libs/wire-api/src/Wire/API/PostgresMarshall.hs b/libs/wire-api/src/Wire/API/PostgresMarshall.hs
index e1a6f55f18d..24ff507ec27 100644
--- a/libs/wire-api/src/Wire/API/PostgresMarshall.hs
+++ b/libs/wire-api/src/Wire/API/PostgresMarshall.hs
@@ -521,6 +521,9 @@ instance (PostgresMarshall a1 b1, PostgresMarshall a2 b2, PostgresMarshall a3 b3
instance (PostgresMarshall a1 b1, PostgresMarshall a2 b2, PostgresMarshall a3 b3, PostgresMarshall a4 b4, PostgresMarshall a5 b5, PostgresMarshall a6 b6, PostgresMarshall a7 b7, PostgresMarshall a8 b8, PostgresMarshall a9 b9, PostgresMarshall a10 b10, PostgresMarshall a11 b11, PostgresMarshall a12 b12, PostgresMarshall a13 b13, PostgresMarshall a14 b14, PostgresMarshall a15 b15, PostgresMarshall a16 b16, PostgresMarshall a17 b17, PostgresMarshall a18 b18, PostgresMarshall a19 b19, PostgresMarshall a20 b20, PostgresMarshall a21 b21, PostgresMarshall a22 b22, PostgresMarshall a23 b23, PostgresMarshall a24 b24) => PostgresMarshall (a1, a2, a3, a4, a5, a6, a7, a8, a9, a10, a11, a12, a13, a14, a15, a16, a17, a18, a19, a20, a21, a22, a23, a24) (b1, b2, b3, b4, b5, b6, b7, b8, b9, b10, b11, b12, b13, b14, b15, b16, b17, b18, b19, b20, b21, b22, b23, b24) where
postgresMarshall (a1, a2, a3, a4, a5, a6, a7, a8, a9, a10, a11, a12, a13, a14, a15, a16, a17, a18, a19, a20, a21, a22, a23, a24) = (postgresMarshall a1, postgresMarshall a2, postgresMarshall a3, postgresMarshall a4, postgresMarshall a5, postgresMarshall a6, postgresMarshall a7, postgresMarshall a8, postgresMarshall a9, postgresMarshall a10, postgresMarshall a11, postgresMarshall a12, postgresMarshall a13, postgresMarshall a14, postgresMarshall a15, postgresMarshall a16, postgresMarshall a17, postgresMarshall a18, postgresMarshall a19, postgresMarshall a20, postgresMarshall a21, postgresMarshall a22, postgresMarshall a23, postgresMarshall a24)
+instance (PostgresMarshall a1 b1, PostgresMarshall a2 b2, PostgresMarshall a3 b3, PostgresMarshall a4 b4, PostgresMarshall a5 b5, PostgresMarshall a6 b6, PostgresMarshall a7 b7, PostgresMarshall a8 b8, PostgresMarshall a9 b9, PostgresMarshall a10 b10, PostgresMarshall a11 b11, PostgresMarshall a12 b12, PostgresMarshall a13 b13, PostgresMarshall a14 b14, PostgresMarshall a15 b15, PostgresMarshall a16 b16, PostgresMarshall a17 b17, PostgresMarshall a18 b18, PostgresMarshall a19 b19, PostgresMarshall a20 b20, PostgresMarshall a21 b21, PostgresMarshall a22 b22, PostgresMarshall a23 b23, PostgresMarshall a24 b24, PostgresMarshall a25 b25) => PostgresMarshall (a1, a2, a3, a4, a5, a6, a7, a8, a9, a10, a11, a12, a13, a14, a15, a16, a17, a18, a19, a20, a21, a22, a23, a24, a25) (b1, b2, b3, b4, b5, b6, b7, b8, b9, b10, b11, b12, b13, b14, b15, b16, b17, b18, b19, b20, b21, b22, b23, b24, b25) where
+ postgresMarshall (a1, a2, a3, a4, a5, a6, a7, a8, a9, a10, a11, a12, a13, a14, a15, a16, a17, a18, a19, a20, a21, a22, a23, a24, a25) = (postgresMarshall a1, postgresMarshall a2, postgresMarshall a3, postgresMarshall a4, postgresMarshall a5, postgresMarshall a6, postgresMarshall a7, postgresMarshall a8, postgresMarshall a9, postgresMarshall a10, postgresMarshall a11, postgresMarshall a12, postgresMarshall a13, postgresMarshall a14, postgresMarshall a15, postgresMarshall a16, postgresMarshall a17, postgresMarshall a18, postgresMarshall a19, postgresMarshall a20, postgresMarshall a21, postgresMarshall a22, postgresMarshall a23, postgresMarshall a24, postgresMarshall a25)
+
instance PostgresMarshall UUID (Id a) where
postgresMarshall = toUUID
@@ -989,6 +992,35 @@ instance (PostgresUnmarshall a1 b1, PostgresUnmarshall a2 b2, PostgresUnmarshall
<*> postgresUnmarshall a23
<*> postgresUnmarshall a24
+instance (PostgresUnmarshall a1 b1, PostgresUnmarshall a2 b2, PostgresUnmarshall a3 b3, PostgresUnmarshall a4 b4, PostgresUnmarshall a5 b5, PostgresUnmarshall a6 b6, PostgresUnmarshall a7 b7, PostgresUnmarshall a8 b8, PostgresUnmarshall a9 b9, PostgresUnmarshall a10 b10, PostgresUnmarshall a11 b11, PostgresUnmarshall a12 b12, PostgresUnmarshall a13 b13, PostgresUnmarshall a14 b14, PostgresUnmarshall a15 b15, PostgresUnmarshall a16 b16, PostgresUnmarshall a17 b17, PostgresUnmarshall a18 b18, PostgresUnmarshall a19 b19, PostgresUnmarshall a20 b20, PostgresUnmarshall a21 b21, PostgresUnmarshall a22 b22, PostgresUnmarshall a23 b23, PostgresUnmarshall a24 b24, PostgresUnmarshall a25 b25) => PostgresUnmarshall (a1, a2, a3, a4, a5, a6, a7, a8, a9, a10, a11, a12, a13, a14, a15, a16, a17, a18, a19, a20, a21, a22, a23, a24, a25) (b1, b2, b3, b4, b5, b6, b7, b8, b9, b10, b11, b12, b13, b14, b15, b16, b17, b18, b19, b20, b21, b22, b23, b24, b25) where
+ postgresUnmarshall (a1, a2, a3, a4, a5, a6, a7, a8, a9, a10, a11, a12, a13, a14, a15, a16, a17, a18, a19, a20, a21, a22, a23, a24, a25) =
+ (,,,,,,,,,,,,,,,,,,,,,,,,)
+ <$> postgresUnmarshall a1
+ <*> postgresUnmarshall a2
+ <*> postgresUnmarshall a3
+ <*> postgresUnmarshall a4
+ <*> postgresUnmarshall a5
+ <*> postgresUnmarshall a6
+ <*> postgresUnmarshall a7
+ <*> postgresUnmarshall a8
+ <*> postgresUnmarshall a9
+ <*> postgresUnmarshall a10
+ <*> postgresUnmarshall a11
+ <*> postgresUnmarshall a12
+ <*> postgresUnmarshall a13
+ <*> postgresUnmarshall a14
+ <*> postgresUnmarshall a15
+ <*> postgresUnmarshall a16
+ <*> postgresUnmarshall a17
+ <*> postgresUnmarshall a18
+ <*> postgresUnmarshall a19
+ <*> postgresUnmarshall a20
+ <*> postgresUnmarshall a21
+ <*> postgresUnmarshall a22
+ <*> postgresUnmarshall a23
+ <*> postgresUnmarshall a24
+ <*> postgresUnmarshall a25
+
instance PostgresUnmarshall UUID (Id a) where
postgresUnmarshall = Right . Id
diff --git a/libs/wire-subsystems/postgres-migrations/20260625104559-add-user-migration-pending-deletes.sql b/libs/wire-subsystems/postgres-migrations/20260625104559-add-user-migration-pending-deletes.sql
new file mode 100644
index 00000000000..21143117419
--- /dev/null
+++ b/libs/wire-subsystems/postgres-migrations/20260625104559-add-user-migration-pending-deletes.sql
@@ -0,0 +1,3 @@
+CREATE TABLE user_migration_pending_deletes (
+ id uuid PRIMARY KEY
+ );
diff --git a/libs/wire-subsystems/src/Wire/CodeStore/Migration.hs b/libs/wire-subsystems/src/Wire/CodeStore/Migration.hs
index e652c5619a3..49062ebe50f 100644
--- a/libs/wire-subsystems/src/Wire/CodeStore/Migration.hs
+++ b/libs/wire-subsystems/src/Wire/CodeStore/Migration.hs
@@ -32,12 +32,12 @@ import Hasql.Pool.Extended qualified as Hasql
import Imports
import Polysemy
import Polysemy.Async
+import Polysemy.AtomicState
import Polysemy.Conc (interpretRace)
import Polysemy.Conc qualified as Conc
import Polysemy.Conc.Effect.Race hiding (Timeout)
import Polysemy.Input
import Polysemy.Resource (Resource, bracket, resourceToIOFinal)
-import Polysemy.State
import Polysemy.TinyLog
import Prometheus qualified
import System.Logger qualified as Log
@@ -53,7 +53,7 @@ import Wire.Sem.Logger (mapLogger)
import Wire.Sem.Logger.TinyLog (loggerToTinyLog)
type EffectStack =
- [ State Int,
+ [ AtomicState Int,
Input ClientState,
Input Hasql.Pool,
Input (Either HttpsUrl (Map Domain HttpsUrl)),
@@ -97,7 +97,7 @@ interpreter cassClient pgPool logger name =
. runInputConst (Right mempty)
. runInputConst pgPool
. runInputConst cassClient
- . runState 0
+ . atomicStateToIO 0
migrateAllCodes ::
( Member (Input Hasql.Pool) r,
@@ -105,7 +105,7 @@ migrateAllCodes ::
Member (Embed IO) r,
Member (Input ClientState) r,
Member TinyLog r,
- Member (State Int) r,
+ Member (AtomicState Int) r,
Member Resource r,
Member Race r
) =>
diff --git a/libs/wire-subsystems/src/Wire/ConversationStore/Migration.hs b/libs/wire-subsystems/src/Wire/ConversationStore/Migration.hs
index 8c5ddc11bd7..60ab456464e 100644
--- a/libs/wire-subsystems/src/Wire/ConversationStore/Migration.hs
+++ b/libs/wire-subsystems/src/Wire/ConversationStore/Migration.hs
@@ -43,11 +43,11 @@ import Hasql.Transaction.Sessions
import Imports
import Polysemy
import Polysemy.Async
+import Polysemy.AtomicState
import Polysemy.Conc hiding (timeout_)
import Polysemy.Error
import Polysemy.Input
import Polysemy.Resource (Resource, resourceToIOFinal)
-import Polysemy.State
import Polysemy.TinyLog
import Prometheus qualified
import System.Logger qualified as Log
@@ -81,7 +81,7 @@ import Wire.StoredConversation
-- * Top level logic
type EffectStack =
- [ State Int,
+ [ AtomicState Int,
Input ClientState,
Input Hasql.Pool,
Resource,
@@ -144,7 +144,7 @@ interpreter cassClient pgPool logger name =
. resourceToIOFinal
. runInputConst pgPool
. runInputConst cassClient
- . runState 0
+ . atomicStateToIO 0
migrateAllConversations ::
( Member (Input Hasql.Pool) r,
@@ -154,7 +154,7 @@ migrateAllConversations ::
Member Async r,
Member Race r,
Member Resource r,
- Member (State Int) r,
+ Member (AtomicState Int) r,
Member (Concurrency Unsafe) r
) =>
MigrationOptions ->
@@ -181,7 +181,7 @@ migrateAllUsers ::
Member Async r,
Member Race r,
Member Resource r,
- Member (State Int) r,
+ Member (AtomicState Int) r,
Member (Concurrency 'Unsafe) r
) =>
MigrationOptions ->
@@ -197,11 +197,11 @@ migrateAllUsers migOpts migCounter migDuration = do
select :: PrepQuery R () (Identity UserId)
select = "select distinct user from user_remote_conv"
-handleErrors :: (Member (State Int) r, Member TinyLog r) => (Id a -> Sem (Error MigrationLockError : Error UsageError : r) b) -> ByteString -> Id a -> Sem r (Maybe b)
+handleErrors :: (Member (AtomicState Int) r, Member TinyLog r) => (Id a -> Sem (Error MigrationLockError : Error UsageError : r) b) -> ByteString -> Id a -> Sem r (Maybe b)
handleErrors action lockType id_ =
join <$> handleError (handleError action lockType) lockType id_
-handleError :: (Member (State Int) r, Member TinyLog r, Show e) => (Id a -> Sem (Error e : r) b) -> ByteString -> Id a -> Sem r (Maybe b)
+handleError :: (Member (AtomicState Int) r, Member TinyLog r, Show e) => (Id a -> Sem (Error e : r) b) -> ByteString -> Id a -> Sem r (Maybe b)
handleError action lockType id_ = do
eithErr <- runError (action id_)
case eithErr of
@@ -211,7 +211,7 @@ handleError action lockType id_ = do
Log.msg (Log.val "error occurred during migration")
. Log.field lockType (idToText id_)
. Log.field "error" (show e)
- modify (+ 1)
+ atomicModify (+ 1)
pure Nothing
-- * Conversations
diff --git a/libs/wire-subsystems/src/Wire/DomainRegistrationStore/Migration.hs b/libs/wire-subsystems/src/Wire/DomainRegistrationStore/Migration.hs
index 5047d4b61e2..bd29da928f4 100644
--- a/libs/wire-subsystems/src/Wire/DomainRegistrationStore/Migration.hs
+++ b/libs/wire-subsystems/src/Wire/DomainRegistrationStore/Migration.hs
@@ -27,17 +27,16 @@ import Data.Conduit.List qualified as C
import Data.Domain
import Data.Id
import Database.CQL.Protocol (Record (asRecord), TupleType)
-import Hasql.Pool (UsageError)
import Hasql.Pool.Extended qualified as Hasql
import Imports hiding (lookup)
import Polysemy
import Polysemy.Async
+import Polysemy.AtomicState
import Polysemy.Conc (interpretRace)
import Polysemy.Conc.Effect.Race hiding (Timeout)
import Polysemy.Error
import Polysemy.Input
import Polysemy.Resource (Resource, resourceToIOFinal)
-import Polysemy.State
import Polysemy.TinyLog
import Prometheus qualified
import System.Logger qualified as Log
@@ -55,7 +54,7 @@ import Wire.Sem.Logger (mapLogger)
import Wire.Sem.Logger.TinyLog (loggerToTinyLog)
type EffectStack =
- [ State Int,
+ [ AtomicState Int,
Input ClientState,
Input Hasql.Pool,
Resource,
@@ -97,14 +96,14 @@ interpreter cassClient pgPool logger name =
. resourceToIOFinal
. runInputConst pgPool
. runInputConst cassClient
- . runState 0
+ . atomicStateToIO 0
migrateAllDomainRegistrations ::
( Member (Input Hasql.Pool) r,
Member (Embed IO) r,
Member (Input ClientState) r,
Member TinyLog r,
- Member (State Int) r,
+ Member (AtomicState Int) r,
Member Async r,
Member Race r,
Member Resource r
@@ -122,7 +121,7 @@ migrateAllDomainRegistrations migOpts migCounter migDuration = do
lift $ info $ Log.msg (Log.val "migrateAllDomainRegistrations")
withCount (paginateSem selectAllRegistrations (paramsP LocalQuorum () migOpts.pageSize) x5)
.| logRetrievedPage migOpts.pageSize asRecord
- .| C.mapM_ (traverse_ (\row -> handleRegistrationErrors (toByteString' (show row.domain)) (migrateDomainRegistrationRow migOpts migCounter migDuration row)))
+ .| C.mapM_ (traverse_ (\row -> handleLockAndDBErrors (toByteString' (show row.domain)) (migrateDomainRegistrationRow migOpts migCounter migDuration row)))
migrateDomainRegistrationRow ::
( PGConstraints r,
@@ -175,24 +174,3 @@ selectAllRegistrations =
selectAllChallenges :: PrepQuery R () (ChallengeId, Domain, Token, DnsVerificationToken, Int32)
selectAllChallenges =
"SELECT id, domain, challenge_token_hash, dns_verification_token, ttl(challenge_token_hash) FROM domain_registration_challenge"
-
-handleRegistrationErrors ::
- ( Member (State Int) r,
- Member TinyLog r
- ) =>
- ByteString ->
- (Sem (Error MigrationLockError : Error UsageError : r) ()) ->
- Sem r ()
-handleRegistrationErrors key action = do
- eithErr <- runError (runError action)
- case eithErr of
- Right (Right _) -> pure ()
- Right (Left e) -> logError (show e)
- Left e -> logError (show e)
- where
- logError e = do
- warn $
- Log.msg (Log.val "error occurred during migration")
- . Log.field "key" (show key)
- . Log.field "error" e
- modify (+ 1)
diff --git a/libs/wire-subsystems/src/Wire/Migration.hs b/libs/wire-subsystems/src/Wire/Migration.hs
index 325910448ba..b15edbfeb47 100644
--- a/libs/wire-subsystems/src/Wire/Migration.hs
+++ b/libs/wire-subsystems/src/Wire/Migration.hs
@@ -32,12 +32,12 @@ import Hasql.Pool qualified as Hasql
import Imports
import Polysemy
import Polysemy.Async
+import Polysemy.AtomicState
import Polysemy.Conc hiding (timeout_)
import Polysemy.Conc qualified as Conc
import Polysemy.Error
import Polysemy.Input
import Polysemy.Resource
-import Polysemy.State
import Polysemy.Time
import Polysemy.TinyLog
import Prometheus qualified
@@ -152,7 +152,7 @@ paginateSem q p r = do
handleErrors ::
forall r.
- ( Member (State Int) r,
+ ( Member (AtomicState Int) r,
Member TinyLog r
) =>
ByteString ->
@@ -167,7 +167,28 @@ handleErrors key action = do
Log.msg (Log.val "error occurred during migration")
. Log.field "key" (show key)
. Log.field "error" (show e)
- modify (+ 1)
+ atomicModify (+ 1)
+
+handleLockAndDBErrors ::
+ ( Member (AtomicState Int) r,
+ Member TinyLog r
+ ) =>
+ ByteString ->
+ (Sem (Error MigrationLockError : Error Hasql.UsageError : r) ()) ->
+ Sem r ()
+handleLockAndDBErrors key action = do
+ eithErr <- runError (runError action)
+ case eithErr of
+ Right (Right _) -> pure ()
+ Right (Left e) -> logError (show e)
+ Left e -> logError (show e)
+ where
+ logError e = do
+ warn $
+ Log.msg (Log.val "error occurred during migration")
+ . Log.field "key" (show key)
+ . Log.field "error" e
+ atomicModify (+ 1)
withExclusiveMigrationLockAndTimeout ::
forall x r.
diff --git a/libs/wire-subsystems/src/Wire/MigrationLock.hs b/libs/wire-subsystems/src/Wire/MigrationLock.hs
index 8c97876170f..9befa82247c 100644
--- a/libs/wire-subsystems/src/Wire/MigrationLock.hs
+++ b/libs/wire-subsystems/src/Wire/MigrationLock.hs
@@ -86,6 +86,8 @@ data MigrationLockError = TimedOutAcquiringLock
instance APIError MigrationLockError where
toResponse = waiErrorToJSONResponse . migrationLockErrorToWai
+instance Exception MigrationLockError
+
migrationLockErrorToHttpError :: MigrationLockError -> HttpError
migrationLockErrorToHttpError = StdError . migrationLockErrorToWai
@@ -117,7 +119,8 @@ withMigrationLocks lockType maxWait lockables action = do
pool <- (.rawPool) <$> input @HasqlPoolExt.Pool
lockThread <- async . embed . Hasql.use pool $ do
- let lockIds = fmap lockKey lockables
+ -- Sort lockIds to avoid deadlocks
+ let lockIds = sort $ fmap lockKey lockables
Session.statement lockIds acquireLocks
liftIO $ putMVar lockAcquired ()
diff --git a/libs/wire-subsystems/src/Wire/TeamFeatureStore/Migration.hs b/libs/wire-subsystems/src/Wire/TeamFeatureStore/Migration.hs
index 1edeb7daf2c..048193e3f16 100644
--- a/libs/wire-subsystems/src/Wire/TeamFeatureStore/Migration.hs
+++ b/libs/wire-subsystems/src/Wire/TeamFeatureStore/Migration.hs
@@ -27,11 +27,11 @@ import Hasql.Pool.Extended qualified as Hasql
import Imports
import Polysemy
import Polysemy.Async
+import Polysemy.AtomicState
import Polysemy.Conc
import Polysemy.Error
import Polysemy.Input
import Polysemy.Resource (Resource, resourceToIOFinal)
-import Polysemy.State
import Polysemy.TinyLog
import Prometheus qualified
import System.Logger qualified as Log
@@ -49,7 +49,7 @@ migrateAllTeamFeatures ::
Member (Embed IO) r,
Member (Input ClientState) r,
Member TinyLog r,
- Member (State Int) r,
+ Member (AtomicState Int) r,
Member Async r,
Member Race r,
Member Resource r
@@ -65,7 +65,7 @@ migrateAllTeamFeatures migOpts migCounter migDuration = do
.| C.mapM_ (traverse_ (\row@(tid, feat, _, _, _) -> handleErrors (toByteString' (idToText tid <> " - " <> feat)) (migrateTeamFeature migOpts migCounter migDuration row)))
type EffectStack =
- [ State Int,
+ [ AtomicState Int,
Input ClientState,
Input Hasql.Pool,
Resource,
@@ -107,7 +107,7 @@ interpreter cassClient pgPool logger name =
. resourceToIOFinal
. runInputConst pgPool
. runInputConst cassClient
- . runState 0
+ . atomicStateToIO 0
migrateTeamFeature ::
( PGConstraints r,
@@ -134,7 +134,7 @@ migrateTeamFeature migOpts migCounter migDuration (tid, name, status, lockStatus
liftIO $ Prometheus.incCounter migCounter
handleErrors ::
- ( Member (State Int) r,
+ ( Member (AtomicState Int) r,
Member TinyLog r
) =>
ByteString ->
@@ -149,10 +149,10 @@ handleErrors key action = do
Log.msg (Log.val "error occurred during migration")
. Log.field "key" (show key)
. Log.field "error" (show e)
- modify (+ 1)
+ atomicModify (+ 1)
Left e -> do
warn $
Log.msg (Log.val "error occurred during migration")
. Log.field "key" (show key)
. Log.field "error" (show e)
- modify (+ 1)
+ atomicModify (+ 1)
diff --git a/libs/wire-subsystems/src/Wire/UserStore/Cassandra.hs b/libs/wire-subsystems/src/Wire/UserStore/Cassandra.hs
index 6226347fece..54fefc2d485 100644
--- a/libs/wire-subsystems/src/Wire/UserStore/Cassandra.hs
+++ b/libs/wire-subsystems/src/Wire/UserStore/Cassandra.hs
@@ -15,27 +15,42 @@
-- You should have received a copy of the GNU Affero General Public License along
-- with this program. If not, see .
-module Wire.UserStore.Cassandra (interpretUserStoreCassandra) where
+module Wire.UserStore.Cassandra
+ ( interpretUserStoreCassandra,
+ interpretUserStoreToCassandraAndPostgres,
+ )
+where
import Cassandra
import Cassandra.Exec (prepared)
import Control.Lens ((^.))
import Data.Handle
import Data.Id
-import Database.CQL.Protocol
+import Data.Map qualified as Map
+import Data.UUID qualified as UUID
+import Database.CQL.Protocol hiding (Error)
import Imports
import Polysemy
+import Polysemy.Async (Async)
+import Polysemy.Conc (Race)
import Polysemy.Embed
import Polysemy.Error
+import Polysemy.Resource (Resource)
+import Polysemy.Time
+import Polysemy.TinyLog (TinyLog)
import Wire.API.Password (Password)
import Wire.API.Provider.Service
import Wire.API.Team.Feature (FeatureStatus)
import Wire.API.User hiding (DeleteUser)
import Wire.API.User.RichInfo
import Wire.API.User.Search (SetSearchable (SetSearchable))
+import Wire.MigrationLock
+import Wire.Postgres (PGConstraints)
import Wire.StoredUser
import Wire.UserStore
+import Wire.UserStore qualified as UserStore
import Wire.UserStore.IndexUser hiding (userId)
+import Wire.UserStore.Postgres (interpretUserStorePostgres)
import Wire.UserStore.Unique
interpretUserStoreCassandra :: (Member (Embed IO) r) => ClientState -> InterpreterFor UserStore r
@@ -79,6 +94,197 @@ interpretUserStoreCassandra casClient =
LookupServiceUsers pid sid mPagingState -> lookupServiceUsersImpl pid sid (paginationStateCassandra =<< mPagingState)
LookupServiceUsersForTeam pid sid tid mPagingState -> lookupServiceUsersForTeamImpl pid sid tid (paginationStateCassandra =<< mPagingState)
+interpretUserStoreToCassandraAndPostgres ::
+ ( PGConstraints r,
+ Member Async r,
+ Member TinyLog r,
+ Member Race r,
+ Member Resource r,
+ Member (Error MigrationLockError) r
+ ) =>
+ ClientState -> InterpreterFor UserStore r
+interpretUserStoreToCassandraAndPostgres casClient =
+ interpret $ \case
+ CreateUser new mbConv -> do
+ -- Store new users in postgresql
+ withMigrationLocks LockShared (MilliSeconds 500) [new.id] $ do
+ isUserInCass <- interpretUserStoreCassandra casClient $ UserStore.doesUserExist new.id
+ if isUserInCass
+ then interpretUserStoreCassandra casClient $ UserStore.createUser new mbConv
+ else interpretUserStorePostgres $ UserStore.createUser new mbConv
+ ActivateUser uid identity ->
+ runAppropriateInterpreter casClient uid $ UserStore.activateUser uid identity
+ DeactivateUser uid ->
+ runAppropriateInterpreter casClient uid $ UserStore.deactivateUser uid
+ GetUsers uids ->
+ withMigrationLocks LockShared (Seconds 2) uids $ do
+ let indexByUserId = foldr (\storedUser -> Map.insert storedUser.id storedUser) Map.empty
+ cassUsers <- indexByUserId <$> interpretUserStoreCassandra casClient (UserStore.getUsers uids)
+ pgUsers <- indexByUserId <$> interpretUserStorePostgres (UserStore.getUsers uids)
+ pure $ mapMaybe (\uid -> Map.lookup uid pgUsers <|> Map.lookup uid cassUsers) uids
+ DoesUserExist uid -> do
+ withMigrationLocks LockShared (MilliSeconds 500) [uid] $ do
+ isUserInPg <- interpretUserStorePostgres $ UserStore.doesUserExist uid
+ if isUserInPg
+ then pure True
+ else interpretUserStoreCassandra casClient $ UserStore.doesUserExist uid
+ GetIndexUser uid ->
+ runAppropriateInterpreter casClient uid $ UserStore.getIndexUser uid
+ GetIndexUsersPaginated pageSize mPagingState -> do
+ paginateOverCassandraAndPostgres
+ (\size state -> interpretUserStoreCassandra casClient $ UserStore.getIndexUsersPaginated size state)
+ (\size state -> interpretUserStorePostgres $ UserStore.getIndexUsersPaginated size state)
+ (PagingExitingUsers $ Id UUID.nil)
+ pageSize
+ mPagingState
+ UpdateUser uid update ->
+ runAppropriateInterpreter casClient uid $ UserStore.updateUser uid update
+ UpdateEmail uid email ->
+ runAppropriateInterpreter casClient uid $ UserStore.updateEmail uid email
+ DeleteEmail uid ->
+ runAppropriateInterpreter casClient uid $ UserStore.deleteEmail uid
+ UpdateEmailUnvalidated uid email ->
+ runAppropriateInterpreter casClient uid $ UserStore.updateEmailUnvalidated uid email
+ DeleteEmailUnvalidated uid ->
+ runAppropriateInterpreter casClient uid $ UserStore.deleteEmailUnvalidated uid
+ LookupName uid ->
+ runAppropriateInterpreter casClient uid $ UserStore.lookupName uid
+ LookupHandle hdl -> do
+ let action = UserStore.lookupHandle hdl
+ interpretUserStorePostgres action >>= \case
+ Nothing -> interpretUserStoreCassandra casClient action
+ Just user -> pure $ Just user
+ GlimpseHandle hdl -> do
+ let action = UserStore.glimpseHandle hdl
+ interpretUserStorePostgres action >>= \case
+ Nothing -> interpretUserStoreCassandra casClient action
+ Just uid -> pure $ Just uid
+ UpdateUserHandleEither uid update -> do
+ -- There is no easy way to handle the race condition that Alice in
+ -- Cassandra and Bob in Postgresql don't claim the same handle. If they
+ -- race to claim a handle, they _can_ both succeed. In this case, the
+ -- migration code _could_ fail to migrate the Alice user, so it has to be
+ -- careful about handling this case.
+ withMigrationLocks LockShared (MilliSeconds 500) [uid] $ do
+ let glimpseAction = UserStore.glimpseHandle update.new
+ cassGlimpse <- interpretUserStoreCassandra casClient glimpseAction
+ pgGlimpse <- interpretUserStorePostgres glimpseAction
+ case (cassGlimpse, pgGlimpse) of
+ (_, Just pgClaimer)
+ | pgClaimer == uid -> pure $ Right ()
+ | otherwise -> pure $ Left StoredUserUpdateHandleExists
+ (Just casClaimer, Nothing)
+ | casClaimer == uid -> pure $ Right ()
+ | otherwise -> pure $ Left StoredUserUpdateHandleExists
+ (Nothing, Nothing) -> do
+ isUserInPg <- interpretUserStorePostgres $ UserStore.doesUserExist uid
+ let action = UserStore.updateUserHandleEither uid update
+ if isUserInPg
+ then interpretUserStorePostgres action
+ else interpretUserStoreCassandra casClient action
+ UpdateSSOId uid ssoId ->
+ runAppropriateInterpreter casClient uid $ UserStore.updateSSOId uid ssoId
+ UpdateManagedBy uid managedBy ->
+ runAppropriateInterpreter casClient uid $ UserStore.updateManagedBy uid managedBy
+ UpdateAccountStatus uid accountStatus ->
+ runAppropriateInterpreter casClient uid $ UserStore.updateAccountStatus uid accountStatus
+ UpdateRichInfo uid richInfo ->
+ runAppropriateInterpreter casClient uid $ UserStore.updateRichInfo uid richInfo
+ UpdateFeatureConferenceCalling uid feat ->
+ runAppropriateInterpreter casClient uid $ UserStore.updateFeatureConferenceCalling uid feat
+ LookupFeatureConferenceCalling uid ->
+ runAppropriateInterpreter casClient uid $ UserStore.lookupFeatureConferenceCalling uid
+ DeleteUser user ->
+ runAppropriateInterpreter casClient (userId user) $ UserStore.deleteUser user
+ LookupStatus uid ->
+ runAppropriateInterpreter casClient uid $ UserStore.lookupStatus uid
+ IsActivated uid ->
+ runAppropriateInterpreter casClient uid $ UserStore.isActivated uid
+ LookupLocale uid ->
+ runAppropriateInterpreter casClient uid $ UserStore.lookupLocale uid
+ GetUserTeam uid ->
+ runAppropriateInterpreter casClient uid $ UserStore.getUserTeam uid
+ UpdateUserTeam uid tid ->
+ runAppropriateInterpreter casClient uid $ UserStore.updateUserTeam uid tid
+ GetRichInfo uid ->
+ runAppropriateInterpreter casClient uid $ UserStore.getRichInfo uid
+ UpsertHashedPassword uid pw ->
+ runAppropriateInterpreter casClient uid $ UserStore.upsertHashedPassword uid pw
+ LookupHashedPassword uid ->
+ runAppropriateInterpreter casClient uid $ UserStore.lookupHashedPassword uid
+ GetUserAuthenticationInfo uid ->
+ runAppropriateInterpreter casClient uid $ UserStore.getUserAuthenticationInfo uid
+ SetUserSearchable uid searchable ->
+ runAppropriateInterpreter casClient uid $ UserStore.setUserSearchable uid searchable
+ DeleteServiceUser pid sid bid ->
+ runAppropriateInterpreter casClient (botUserId bid) $ UserStore.deleteServiceUser pid sid bid
+ LookupServiceUsers pid sid mPagingState ->
+ -- Ignoring the size paramter here makes us potentially return upto 199
+ -- bots instead of 100, but this is ok as this is temporary and the
+ -- callers are not doing anything wrong with a longer list.
+ paginateOverCassandraAndPostgres
+ (\_size state -> interpretUserStoreCassandra casClient $ UserStore.lookupServiceUsers pid sid state)
+ (\_size state -> interpretUserStorePostgres $ UserStore.lookupServiceUsers pid sid state)
+ (BotId $ Id UUID.nil)
+ 100
+ mPagingState
+ LookupServiceUsersForTeam pid sid tid mPagingState ->
+ -- Ignoring the size paramter here makes us potentially return upto 199
+ -- bots instead of 100, but this is ok as this is temporary and the
+ -- callers are not doing anything wrong with a longer list.
+ paginateOverCassandraAndPostgres
+ (\_size state -> interpretUserStoreCassandra casClient $ UserStore.lookupServiceUsersForTeam pid sid tid state)
+ (\_size state -> interpretUserStorePostgres $ UserStore.lookupServiceUsersForTeam pid sid tid state)
+ (BotId $ Id UUID.nil)
+ 100
+ mPagingState
+
+runAppropriateInterpreter ::
+ ( PGConstraints r,
+ Member TinyLog r,
+ Member (Error MigrationLockError) r,
+ Member Async r,
+ Member Race r,
+ Member Resource r
+ ) =>
+ ClientState -> UserId -> InterpreterFor UserStore r
+runAppropriateInterpreter casClient uid action =
+ withMigrationLocks LockShared (MilliSeconds 500) [uid] $ do
+ isUserInPg <- interpretUserStorePostgres $ UserStore.doesUserExist uid
+ if isUserInPg
+ then interpretUserStorePostgres action
+ else interpretUserStoreCassandra casClient action
+
+paginateOverCassandraAndPostgres ::
+ (Int32 -> Maybe (GeneralPaginationState pgMarker) -> Sem r (PageWithState pgMarker pageItem)) ->
+ (Int32 -> Maybe (GeneralPaginationState pgMarker) -> Sem r (PageWithState pgMarker pageItem)) ->
+ pgMarker ->
+ Int32 ->
+ Maybe (GeneralPaginationState pgMarker) ->
+ Sem r (PageWithState pgMarker pageItem)
+paginateOverCassandraAndPostgres getCasPage getPgPage pgStartingMarker pageSize mPagingState = do
+ let getPageFromCassandra = do
+ casPage <- getCasPage pageSize mPagingState
+ if pwsHasMore casPage
+ then pure casPage
+ else do
+ let casSize = fromIntegral (length casPage.pwsResults)
+ remainingSize = pageSize - casSize
+ if remainingSize > 0
+ then do
+ pgPage <- getPageFromPostgres remainingSize Nothing
+ pure
+ PageWithState
+ { pwsResults = casPage.pwsResults <> pgPage.pwsResults,
+ pwsState = pgPage.pwsState
+ }
+ else pure $ casPage {pwsState = Just (PaginationStatePostgres pgStartingMarker)}
+ getPageFromPostgres remainingSize mPgMarker =
+ getPgPage remainingSize (PaginationStatePostgres <$> mPgMarker)
+ case mPagingState of
+ Just (PaginationStatePostgres pgMarker) -> getPageFromPostgres pageSize (Just pgMarker)
+ _ -> getPageFromCassandra
+
createUserImpl :: NewStoredUser -> Maybe (ConvId, Maybe TeamId) -> Client ()
createUserImpl new mbConv = retry x5 . batch $ do
setType BatchLogged
diff --git a/libs/wire-subsystems/src/Wire/UserStore/Migration.hs b/libs/wire-subsystems/src/Wire/UserStore/Migration.hs
new file mode 100644
index 00000000000..ebfe29541d8
--- /dev/null
+++ b/libs/wire-subsystems/src/Wire/UserStore/Migration.hs
@@ -0,0 +1,379 @@
+{-# LANGUAGE RecordWildCards #-}
+{-# OPTIONS_GHC -Wno-ambiguous-fields #-}
+
+-- This file is part of the Wire Server implementation.
+--
+-- Copyright (C) 2026 Wire Swiss GmbH
+--
+-- 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 .
+
+module Wire.UserStore.Migration where
+
+import Cassandra hiding (Set)
+import Cassandra.Util
+import Conduit
+import Data.Conduit.List qualified as C
+import Data.Handle
+import Data.Id
+import Data.Json.Util (UTCTimeMillis)
+import Data.Misc
+import Data.Time
+import Database.CQL.Protocol (Record (..), TupleType)
+import Hasql.Pool.Extended
+import Hasql.Statement qualified as Hasql
+import Hasql.TH (resultlessStatement, singletonStatement)
+import Hasql.Transaction qualified as Transaction
+import Hasql.Transaction.Sessions (IsolationLevel (..), Mode (..))
+import Imports
+import Polysemy
+import Polysemy.Async
+import Polysemy.AtomicState
+import Polysemy.Conc
+import Polysemy.Error
+import Polysemy.Input
+import Polysemy.Resource
+import Polysemy.TinyLog
+import Prometheus qualified
+import System.Logger.Class qualified as Log
+import Wire.API.Password
+import Wire.API.PostgresMarshall
+import Wire.API.User
+import Wire.API.User.RichInfo
+import Wire.Migration
+import Wire.MigrationLock
+import Wire.Postgres
+import Wire.Sem.Concurrency
+import Wire.Sem.Concurrency.IO (unsafelyPerformConcurrency)
+import Wire.Sem.Logger
+import Wire.Sem.Logger.TinyLog (loggerToTinyLog)
+import Wire.UserStore.Migration.Types
+import Wire.UserStore.Postgres
+
+migrateUsersLoop ::
+ MigrationOptions ->
+ ClientState ->
+ Pool ->
+ Log.Logger ->
+ Prometheus.Counter ->
+ Prometheus.Counter ->
+ Prometheus.Counter ->
+ Prometheus.Vector Text Prometheus.Histogram ->
+ IO ()
+migrateUsersLoop migOpts cassClient pgPool logger migCounter migFinished migFailed migDuration =
+ migrationLoop
+ logger
+ "users"
+ migFinished
+ migFailed
+ (interpreter cassClient pgPool logger "users")
+ (migrateAllUsers migOpts migCounter migDuration)
+
+type EffectStack =
+ [ AtomicState Int,
+ Input ClientState,
+ Input Pool,
+ Resource,
+ Async,
+ Race,
+ TinyLog,
+ Embed IO,
+ Concurrency 'Unsafe,
+ Final IO
+ ]
+
+interpreter :: ClientState -> Pool -> Log.Logger -> ByteString -> Sem EffectStack a -> IO (Int, a)
+interpreter cassClient pgPool logger name =
+ runFinal
+ . unsafelyPerformConcurrency
+ . embedToFinal
+ . loggerToTinyLog logger
+ . mapLogger (Log.field "migration" name .)
+ . raiseUnder
+ . interpretRace
+ . asyncToIOFinal
+ . resourceToIOFinal
+ . runInputConst pgPool
+ . runInputConst cassClient
+ . atomicStateToIO 0
+
+migrateAllUsers ::
+ ( Member TinyLog r,
+ Member (Input ClientState) r,
+ Member (Embed IO) r,
+ Member (AtomicState Int) r,
+ Member (Concurrency Unsafe) r,
+ Member (Input Pool) r,
+ Member Async r,
+ Member Race r,
+ Member Resource r
+ ) =>
+ MigrationOptions -> Prometheus.Counter -> Prometheus.Vector Text Prometheus.Histogram -> ConduitM () Void (Sem r) ()
+migrateAllUsers migOpts migCounter migDuration = do
+ lift $ info $ Log.msg (Log.val "migrateAllUsers")
+ withCount (paginateSem select (paramsP LocalQuorum () migOpts.pageSize) x5)
+ .| logRetrievedPage migOpts.pageSize runIdentity
+ .| C.mapM_ (unsafePooledMapConcurrentlyN_ migOpts.parallelism (\uid -> handleLockAndDBErrors "user" (migrateUser migOpts.timeout migCounter migDuration uid)))
+ where
+ select :: PrepQuery R () (Identity UserId)
+ select = "select id from user"
+
+migrateUser ::
+ ( PGConstraints r,
+ Member TinyLog r,
+ Member (Error MigrationLockError) r,
+ Member Async r,
+ Member Race r,
+ Member Resource r,
+ Member (Input ClientState) r
+ ) =>
+ Duration -> Prometheus.Counter -> Prometheus.Vector Text Prometheus.Histogram -> UserId -> Sem r ()
+migrateUser migTimeout migCounter migDuration uid =
+ withExclusiveMigrationLockAndTimeout migTimeout migDuration [uid] $ do
+ cState <- input
+ mCassData <- runClient cState $ getUserData uid
+ case mCassData of
+ Nothing -> pure ()
+ Just cassData -> do
+ let eithPGRow = mkUserRowPG cassData.id cassData.user cassData.handleClaimValidity cassData.richInfo
+ case eithPGRow of
+ Left e ->
+ warn $
+ Log.msg (Log.val "Invalid user found, skipping")
+ . Log.field "id" (idToText cassData.id)
+ . Log.field "error" (show e)
+ Right pgRow -> do
+ case cassData.handleClaimValidity of
+ HandleClaimValid -> pure ()
+ HandleNotClaimed ->
+ info $
+ Log.msg (Log.val "This user has a handle which is not 'claimed' by anyone, this user will lose their handle")
+ . Log.field "user" (idToText pgRow.id_)
+ . Log.field "handle" (show $ fromHandle <$> cassData.user.handle)
+ HandleClaimedByAnotherUser claimedBy -> do
+ warn $
+ Log.msg (Log.val "This user has a handle claimed by someone else, this user will lose their handle")
+ . Log.field "user" (idToText pgRow.id_)
+ . Log.field "handle" (show $ fromHandle <$> cassData.user.handle)
+ . Log.field "legitimate_claim_by" (idToText claimedBy)
+ saveToPostgres pgRow cassData.serviceConv
+ let mServiceTeam = (.teamId) =<< cassData.serviceConv
+ runClient cState $ deleteFromCassandra pgRow.id_ pgRow.handle ((,,mServiceTeam) <$> pgRow.providerId <*> pgRow.serviceId)
+ markDeletionComplete pgRow.id_
+ liftIO $ Prometheus.incCounter migCounter
+
+getUserData :: UserId -> Client (Maybe RawUserData)
+getUserData uid = do
+ mUserRow <- asRecord <$$> query1 selectUserRow (params LocalQuorum (Identity uid))
+ case mUserRow of
+ Nothing -> pure Nothing
+ Just user -> do
+ serviceConv <- case (,) <$> user.providerId <*> user.serviceId of
+ Nothing -> pure Nothing
+ Just (pid, sid) -> asRecord <$$> query1 selectServiceConv (params LocalQuorum (pid, sid, uid))
+ handleClaimValidity <- case user.handle of
+ Nothing -> pure HandleClaimValid
+ Just h -> do
+ mClaimedBy <- runIdentity <$$> query1 selectHandleClaim (params LocalQuorum (Identity h))
+ case mClaimedBy of
+ Nothing -> pure HandleNotClaimed
+ Just claimedBy
+ | claimedBy == uid -> pure HandleClaimValid
+ | otherwise -> pure $ HandleClaimedByAnotherUser claimedBy
+ richInfo <- runIdentity <$$> query1 selectRichInfo (params LocalQuorum (Identity uid))
+ pure $ Just RawUserData {id = uid, ..}
+ where
+ selectUserRow :: PrepQuery R (Identity UserId) (TupleType UserRowCass)
+ selectUserRow =
+ "SELECT accent_id, activated, country, email, email_unvalidated,\
+ \expires, feature_conference_calling, handle, language, managed_by, \
+ \name, password, provider, searchable, service,\
+ \sso_id, status, supported_protocols, team, text_status,\
+ \user_type, assets, picture, writetime(activated)\
+ \FROM user WHERE id = ?"
+
+ selectServiceConv :: PrepQuery R (ProviderId, ServiceId, UserId) (TupleType ServiceConv)
+ selectServiceConv = "SELECT conv, team FROM service_user WHERE provider = ? AND service = ? AND user = ?"
+
+ selectHandleClaim :: PrepQuery R (Identity Handle) (Identity UserId)
+ selectHandleClaim = "SELECT user FROM user_handle WHERE handle = ?"
+
+ selectRichInfo :: PrepQuery R (Identity UserId) (Identity RichInfoAssocList)
+ selectRichInfo = "SELECT json FROM rich_info where user = ?"
+
+data InvalidUserError = UserHasNoName | UserHasNoActivated
+ deriving (Show)
+
+mkUserRowPG :: UserId -> UserRowCass -> HandleClaimValidity -> Maybe RichInfoAssocList -> Either InvalidUserError UserRowPG
+mkUserRowPG id_ cass@UserRowCass {..} handleClaimValidity richInfo = run . runError $ do
+ pgName <- note UserHasNoName cass.name
+ pgActivated <- note UserHasNoActivated cass.activated
+ createdAt <- note UserHasNoActivated $ writetimeToUTC <$> cass.activatedWriteTime
+ pure $
+ UserRowPG
+ { accentId = fromMaybe defaultAccentId cass.accentId,
+ userType = fromMaybe UserTypeRegular cass.userType,
+ name = pgName,
+ activated = pgActivated,
+ handle = case handleClaimValidity of
+ HandleClaimValid -> cass.handle
+ HandleNotClaimed ->
+ -- In this case if we just give this handle to the current user,
+ -- there could be other users with the same situation. We cannot tie
+ -- break here, so we just take away the handle from all users
+ Nothing
+ HandleClaimedByAnotherUser _ ->
+ -- Handle is claimed by someone else, so this user cannot get to
+ -- keep it.
+ Nothing,
+ ..
+ }
+
+{- ORMOLU_DISABLE -}
+type UserTuplePG =
+ (UserId, ColourId, Bool, Maybe Country, Maybe EmailAddress,
+ Maybe EmailAddress, Maybe UTCTimeMillis, Maybe Int32, Maybe Handle, Maybe Language,
+ Maybe ManagedBy, Name, Maybe Password, Maybe ProviderId, Maybe ServiceId,
+ Maybe UserSSOId, Maybe AccountStatus, Maybe (Set BaseProtocolTag), Maybe TeamId, Maybe TextStatus,
+ UserType, Maybe Pict, Maybe RichInfoAssocList, Maybe Bool, UTCTime
+ )
+
+userRowPGToTuple :: UserRowPG -> UserTuplePG
+userRowPGToTuple user =
+ (user.id_, user.accentId, user.activated, user.country,user.email,
+ user.emailUnvalidated, user.expires, user.featureConferenceCalling, user.handle, user.language,
+ user.managedBy, user.name, user.password, user.providerId, user.serviceId,
+ user.ssoId, user.status, user.supportedProtocols, user.teamId, user.textStatus,
+ user.userType, user.pict, user.richInfo, user.searchable, user.createdAt)
+{- ORMOLU_ENABLE -}
+
+saveToPostgres :: (PGConstraints r, Member TinyLog r) => UserRowPG -> Maybe ServiceConv -> Sem r ()
+saveToPostgres user mServiceConv = do
+ isHandleRemoved <- runTransactionWithRetry Serializable Write $ do
+ isHandleRemoved <- case user.status of
+ -- bots are deleted by just updating their status to deleted and deleting
+ -- the rows in service_user and service_team tables.
+ Just Deleted
+ | user.userType /= UserTypeBot -> do
+ Transaction.statement (user.id_, user.teamId, user.createdAt) insertDeleted
+ pure False
+ _ -> do
+ removeHandle <-
+ maybe
+ (pure False)
+ (\h -> Transaction.statement (user.id_, h) isHandleTaken)
+ user.handle
+ let userTuple =
+ userRowPGToTuple $
+ if removeHandle
+ then user {handle = Nothing}
+ else user
+ Transaction.statement userTuple insertUser
+ for_ user.assets $ \assets -> do
+ Transaction.statement user.id_ deleteAssetsStatement
+ Transaction.statement (mkAssetRows user.id_ assets) insertAssetsStatement
+ when (user.status /= Just Deleted) $ do
+ for_ mServiceConv $ \serviceConv ->
+ Transaction.statement (user.id_, serviceConv.convId, serviceConv.teamId) insertBotConv
+ pure removeHandle
+ Transaction.statement user.id_ markPendingDelete
+ pure isHandleRemoved
+
+ when isHandleRemoved . warn $
+ Log.msg (Log.val "Duplicate handle claim found, this user doesn't have a handle anymore")
+ . Log.field "user" (idToText user.id_)
+ . Log.field "handle" (show $ fromHandle <$> user.handle)
+ where
+ isHandleTaken :: Hasql.Statement (UserId, Handle) Bool
+ isHandleTaken =
+ dimapPG
+ [singletonStatement|
+ SELECT EXISTS (SELECT 1 FROM wire_user where handle = $2 :: text AND id != $1 :: uuid) :: bool
+ |]
+ insertDeleted :: Hasql.Statement (UserId, Maybe TeamId, UTCTime) ()
+ insertDeleted =
+ lmapPG
+ [resultlessStatement|
+ INSERT INTO deleted_user
+ (id, team, created_at)
+ VALUES ($1 :: uuid, $2 :: uuid?, $3 :: timestamptz)
+ ON CONFLICT (id) DO NOTHING
+ |]
+ insertUser :: Hasql.Statement UserTuplePG ()
+ insertUser =
+ lmapPG
+ [resultlessStatement|
+ INSERT INTO wire_user
+ (id, accent_id, activated, country, email,
+ email_unvalidated, expires, feature_conference_calling, handle, language,
+ managed_by, name, password, provider, service,
+ sso_id, account_status, supported_protocols, team, text_status,
+ user_type, picture, rich_info, searchable, created_at
+ )
+ VALUES
+ ($1 :: uuid, $2 :: integer, $3 :: boolean, $4 :: text?, $5 :: text?,
+ $6 :: text?, $7 :: timestamptz?, $8 :: integer?, $9 :: text?, $10 :: text?,
+ $11 :: integer?, $12 :: text, $13 :: text?, $14 :: uuid?, $15 :: uuid?,
+ $16 :: jsonb?, $17 :: integer?, $18 :: integer?, $19 :: uuid?, $20 :: text?,
+ $21 :: integer, $22 :: jsonb?, $23 :: jsonb?, $24 :: boolean?, $25 :: timestamptz
+ )
+ ON CONFLICT (id) DO NOTHING
+ |]
+ insertBotConv :: Hasql.Statement (UserId, ConvId, Maybe TeamId) ()
+ insertBotConv =
+ lmapPG
+ [resultlessStatement|
+ INSERT INTO bot_conv
+ (id, conv, conv_team)
+ VALUES ($1 :: uuid, $2 :: uuid, $3 :: uuid?)
+ |]
+
+ markPendingDelete :: Hasql.Statement UserId ()
+ markPendingDelete =
+ lmapPG
+ [resultlessStatement|
+ INSERT INTO user_migration_pending_deletes (id)
+ VALUES ($1 :: uuid)
+ ON CONFLICT (id) DO NOTHING
+ |]
+
+markDeletionComplete :: (PGConstraints r) => UserId -> Sem r ()
+markDeletionComplete uid =
+ runStatement uid stmt
+ where
+ stmt :: Hasql.Statement UserId ()
+ stmt = lmapPG [resultlessStatement|DELETE FROM user_migration_pending_deletes WHERE id = $1 :: uuid|]
+
+deleteFromCassandra :: UserId -> Maybe Handle -> Maybe (ProviderId, ServiceId, Maybe TeamId) -> Client ()
+deleteFromCassandra uid mHandle mService = do
+ for_ mHandle $ \handle -> write deleteHandle (params LocalQuorum (Identity handle))
+ for_ mService $ \(pid, sid, mTid) -> do
+ write deleteServiceUser (params LocalQuorum (pid, sid, uid))
+ for_ mTid $ \tid -> write deleteServiceTeam (params LocalQuorum (pid, sid, tid, uid))
+ write deleteRichInfo (params LocalQuorum (Identity uid))
+ write deleteUser (params LocalQuorum (Identity uid))
+ where
+ deleteUser :: PrepQuery W (Identity UserId) ()
+ deleteUser = "DELETE FROM user WHERE id = ?"
+
+ deleteHandle :: PrepQuery W (Identity Handle) ()
+ deleteHandle = "DELETE FROM user_handle WHERE handle = ?"
+
+ deleteServiceUser :: PrepQuery W (ProviderId, ServiceId, UserId) ()
+ deleteServiceUser = "DELETE FROM service_user WHERE provider = ? AND service = ? AND user = ?"
+
+ deleteServiceTeam :: PrepQuery W (ProviderId, ServiceId, TeamId, UserId) ()
+ deleteServiceTeam = "DELETE FROM service_team WHERE provider = ? AND service = ? AND team = ? AND user = ?"
+
+ deleteRichInfo :: PrepQuery W (Identity UserId) ()
+ deleteRichInfo = "DELETE FROM rich_info WHERE user = ?"
diff --git a/libs/wire-subsystems/src/Wire/UserStore/Migration/Types.hs b/libs/wire-subsystems/src/Wire/UserStore/Migration/Types.hs
new file mode 100644
index 00000000000..07619adc6c5
--- /dev/null
+++ b/libs/wire-subsystems/src/Wire/UserStore/Migration/Types.hs
@@ -0,0 +1,111 @@
+{-# LANGUAGE TemplateHaskell #-}
+
+-- This file is part of the Wire Server implementation.
+--
+-- Copyright (C) 2026 Wire Swiss GmbH
+--
+-- 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 .
+
+module Wire.UserStore.Migration.Types where
+
+import Cassandra.Util
+import Data.Handle
+import Data.Id
+import Data.Json.Util
+import Data.Time
+import Database.CQL.Protocol (Record (..), TupleType, recordInstance)
+import Imports
+import Wire.API.Password
+import Wire.API.User
+import Wire.API.User.RichInfo
+
+data RawUserData = RawUserData
+ { id :: UserId,
+ user :: UserRowCass,
+ richInfo :: Maybe RichInfoAssocList,
+ serviceConv :: Maybe ServiceConv,
+ handleClaimValidity :: HandleClaimValidity
+ }
+
+data HandleClaimValidity
+ = HandleClaimValid
+ | HandleNotClaimed
+ | HandleClaimedByAnotherUser UserId
+
+-- | Some fields are read as 'Maybe' even if they're supposed to always be
+-- there. This is to deal with potential old data in the DB.
+data UserRowCass = UserRowCass
+ { accentId :: Maybe ColourId,
+ activated :: Maybe Bool,
+ country :: Maybe Country,
+ email :: Maybe EmailAddress,
+ emailUnvalidated :: Maybe EmailAddress,
+ expires :: Maybe UTCTimeMillis,
+ featureConferenceCalling :: Maybe Int32,
+ handle :: Maybe Handle,
+ language :: Maybe Language,
+ managedBy :: Maybe ManagedBy,
+ name :: Maybe Name,
+ password :: Maybe Password,
+ providerId :: Maybe ProviderId,
+ searchable :: Maybe Bool,
+ serviceId :: Maybe ServiceId,
+ ssoId :: Maybe UserSSOId,
+ status :: Maybe AccountStatus,
+ supportedProtocols :: Maybe (Set BaseProtocolTag),
+ teamId :: Maybe TeamId,
+ textStatus :: Maybe TextStatus,
+ userType :: Maybe UserType,
+ assets :: Maybe [Asset],
+ pict :: Maybe Pict,
+ activatedWriteTime :: Maybe (Writetime ())
+ }
+
+data ServiceConv = ServiceConv
+ { convId :: ConvId,
+ teamId :: Maybe TeamId
+ }
+
+data UserRowPG = UserRowPG
+ { id_ :: UserId,
+ accentId :: ColourId,
+ activated :: Bool,
+ country :: Maybe Country,
+ email :: Maybe EmailAddress,
+ emailUnvalidated :: Maybe EmailAddress,
+ expires :: Maybe UTCTimeMillis,
+ featureConferenceCalling :: Maybe Int32,
+ handle :: Maybe Handle,
+ language :: Maybe Language,
+ managedBy :: Maybe ManagedBy,
+ name :: Name,
+ password :: Maybe Password,
+ providerId :: Maybe ProviderId,
+ searchable :: Maybe Bool,
+ serviceId :: Maybe ServiceId,
+ ssoId :: Maybe UserSSOId,
+ status :: Maybe AccountStatus,
+ supportedProtocols :: Maybe (Set BaseProtocolTag),
+ teamId :: Maybe TeamId,
+ textStatus :: Maybe TextStatus,
+ userType :: UserType,
+ assets :: Maybe [Asset],
+ pict :: Maybe Pict,
+ richInfo :: Maybe RichInfoAssocList,
+ createdAt :: UTCTime
+ }
+
+recordInstance ''UserRowCass
+
+recordInstance ''ServiceConv
diff --git a/libs/wire-subsystems/src/Wire/UserStore/Postgres.hs b/libs/wire-subsystems/src/Wire/UserStore/Postgres.hs
index 88177b95cde..ca9e002bfc5 100644
--- a/libs/wire-subsystems/src/Wire/UserStore/Postgres.hs
+++ b/libs/wire-subsystems/src/Wire/UserStore/Postgres.hs
@@ -18,7 +18,13 @@
-- You should have received a copy of the GNU Affero General Public License along
-- with this program. If not, see .
-module Wire.UserStore.Postgres (interpretUserStorePostgres) where
+module Wire.UserStore.Postgres
+ ( interpretUserStorePostgres,
+ deleteAssetsStatement,
+ insertAssetsStatement,
+ mkAssetRows,
+ )
+where
import Cassandra (GeneralPaginationState (PaginationStatePostgres), PageWithState (..), paginationStatePostgres)
import Data.Handle
diff --git a/libs/wire-subsystems/wire-subsystems.cabal b/libs/wire-subsystems/wire-subsystems.cabal
index 42196d6e18c..abfa57d8442 100644
--- a/libs/wire-subsystems/wire-subsystems.cabal
+++ b/libs/wire-subsystems/wire-subsystems.cabal
@@ -501,6 +501,8 @@ library
Wire.UserStore
Wire.UserStore.Cassandra
Wire.UserStore.IndexUser
+ Wire.UserStore.Migration
+ Wire.UserStore.Migration.Types
Wire.UserStore.Postgres
Wire.UserStore.Unique
Wire.UserSubsystem
diff --git a/postgres-schema.sql b/postgres-schema.sql
index b2d1587ab49..de4a57a0f2e 100644
--- a/postgres-schema.sql
+++ b/postgres-schema.sql
@@ -1614,6 +1614,17 @@ CREATE TABLE public.user_group_member (
ALTER TABLE public.user_group_member OWNER TO "wire-server";
+--
+-- Name: user_migration_pending_deletes; Type: TABLE; Schema: public; Owner: wire-server
+--
+
+CREATE TABLE public.user_migration_pending_deletes (
+ id uuid NOT NULL
+);
+
+
+ALTER TABLE public.user_migration_pending_deletes OWNER TO "wire-server";
+
--
-- Name: wire_user; Type: TABLE; Schema: public; Owner: wire-server
--
@@ -2012,6 +2023,14 @@ ALTER TABLE ONLY public.user_group
ADD CONSTRAINT user_group_pkey PRIMARY KEY (team_id, id);
+--
+-- Name: user_migration_pending_deletes user_migration_pending_deletes_pkey; Type: CONSTRAINT; Schema: public; Owner: wire-server
+--
+
+ALTER TABLE ONLY public.user_migration_pending_deletes
+ ADD CONSTRAINT user_migration_pending_deletes_pkey PRIMARY KEY (id);
+
+
--
-- Name: wire_user wire_user_handle_key; Type: CONSTRAINT; Schema: public; Owner: wire-server
--
diff --git a/services/background-worker/background-worker.integration.yaml b/services/background-worker/background-worker.integration.yaml
index e264ce14016..b0bd0d172e4 100644
--- a/services/background-worker/background-worker.integration.yaml
+++ b/services/background-worker/background-worker.integration.yaml
@@ -58,6 +58,7 @@ migrationOptions:
migrateConversationCodes: false
migrateTeamFeatures: false
migrateDomainRegistration: false
+migrateUsers: false
# Background jobs consumer configuration for integration
backgroundJobs:
diff --git a/services/background-worker/src/Wire/BackgroundWorker.hs b/services/background-worker/src/Wire/BackgroundWorker.hs
index b57ba12df40..6c12b02e816 100644
--- a/services/background-worker/src/Wire/BackgroundWorker.hs
+++ b/services/background-worker/src/Wire/BackgroundWorker.hs
@@ -78,6 +78,14 @@ run opts galleyOpts = do
withNamedLogger "migrate-domain-registration" $
Migrations.domainRegistration opts.migrationOptions
else pure $ pure ()
+ cleanupUsersMigration <-
+ if opts.migrateUsers
+ then
+ runAppT env $
+ withNamedLogger "migrate-users" $
+ Migrations.users opts.migrationOptions
+ else pure $ pure ()
+
cleanupJobs <-
runAppT env $
withNamedLogger "background-job-consumer" $
@@ -89,13 +97,14 @@ run opts galleyOpts = do
let cleanup =
void $
runConcurrently $
- (,,,,,,,)
+ (,,,,,,,,)
<$> Concurrently cleanupDeadUserNotifWatcher
<*> Concurrently cleanupBackendNotifPusher
<*> Concurrently cleanupConvMigration
<*> Concurrently cleanUpConvCodesMigration
<*> Concurrently cleanupTeamFeaturesMigration
<*> Concurrently cleanupDomainRegistrationMigration
+ <*> Concurrently cleanupUsersMigration
<*> Concurrently cleanupJobRunner
<*> Concurrently cleanupJobs
diff --git a/services/background-worker/src/Wire/BackgroundWorker/Options.hs b/services/background-worker/src/Wire/BackgroundWorker/Options.hs
index 61df5d5d14f..035460cfc32 100644
--- a/services/background-worker/src/Wire/BackgroundWorker/Options.hs
+++ b/services/background-worker/src/Wire/BackgroundWorker/Options.hs
@@ -55,6 +55,7 @@ data Opts = Opts
migrateConversationCodes :: !Bool,
migrateTeamFeatures :: !Bool,
migrateDomainRegistration :: !Bool,
+ migrateUsers :: !Bool,
jobs :: JobConfig,
meetingsCleanup :: MeetingsCleanupConfig,
backgroundJobs :: BackgroundJobsConfig
diff --git a/services/background-worker/src/Wire/PostgresMigrations.hs b/services/background-worker/src/Wire/PostgresMigrations.hs
index 604cab0140c..28c6a789a4a 100644
--- a/services/background-worker/src/Wire/PostgresMigrations.hs
+++ b/services/background-worker/src/Wire/PostgresMigrations.hs
@@ -24,10 +24,11 @@ import UnliftIO
import Wire.BackgroundWorker.Env
import Wire.BackgroundWorker.Util
import Wire.CodeStore.Migration
-import Wire.ConversationStore.Migration
+import Wire.ConversationStore.Migration qualified as ConversationStore
import Wire.DomainRegistrationStore.Migration
import Wire.Migration (MigrationOptions)
import Wire.TeamFeatureStore.Migration
+import Wire.UserStore.Migration qualified as UserStore
conversations :: MigrationOptions -> AppT IO CleanupAction
conversations migOpts = do
@@ -45,8 +46,8 @@ conversations migOpts = do
userMigFailed <- register $ counter $ Prometheus.Info "wire_user_remote_convs_migration_failed" "Whether the migration of remote conversation membership data to Postgresql has failed"
userMigDuration <- register $ vector "outcome" $ histogram (Prometheus.Info "wire_user_remote_convs_migration_duration_seconds" "Duration of remote conversation membership migration attempts") defaultBuckets
- convLoop <- async . lift $ migrateConvsLoop migOpts cassClient pgPool logger convMigCounter convMigFinished convMigFailed convMigDuration
- userLoop <- async . lift $ migrateUsersLoop migOpts cassClient pgPool logger userMigCounter userMigFinished userMigFailed userMigDuration
+ convLoop <- async . lift $ ConversationStore.migrateConvsLoop migOpts cassClient pgPool logger convMigCounter convMigFinished convMigFailed convMigDuration
+ userLoop <- async . lift $ ConversationStore.migrateUsersLoop migOpts cassClient pgPool logger userMigCounter userMigFinished userMigFailed userMigDuration
Log.info logger $ Log.msg (Log.val "started conversation migration")
pure $ do
@@ -107,3 +108,21 @@ domainRegistration migOpts = do
pure $ do
Log.info logger $ Log.msg (Log.val "cancelling domain registration migration")
cancel migrationLoop
+
+users :: MigrationOptions -> AppT IO CleanupAction
+users migOpts = do
+ cassClient <- asks (.cassandraBrig)
+ pgPool <- asks (.hasqlPool)
+ logger <- asks (.logger)
+ Log.info logger $ Log.msg (Log.val "starting user migration")
+ count <- register $ counter $ Prometheus.Info "wire_users_migrated_to_pg" "Number of user rows migrated to Postgresql"
+ finished <- register $ counter $ Prometheus.Info "wire_users_migration_finished" "Whether the user migration to Postgresql is finished successfully"
+ failed <- register $ counter $ Prometheus.Info "wire_users_migration_failed" "Whether the user migration to Postgresql has failed"
+ duration <- register $ vector "outcome" $ histogram (Prometheus.Info "wire_users_migration_duration_seconds" "Duration of user migration attempts") defaultBuckets
+
+ migrationLoop <- async . lift $ UserStore.migrateUsersLoop migOpts cassClient pgPool logger count finished failed duration
+
+ Log.info logger $ Log.msg (Log.val "started user migration")
+ pure $ do
+ Log.info logger $ Log.msg (Log.val "cancelling user migration")
+ cancel migrationLoop
diff --git a/services/brig/brig.cabal b/services/brig/brig.cabal
index 89b1d88f7a7..794b137836c 100644
--- a/services/brig/brig.cabal
+++ b/services/brig/brig.cabal
@@ -183,6 +183,7 @@ library
Brig.Schema.V91_UpdateDomainRegistrationSchema_AddWebappUrl
Brig.Schema.V92_AddUserType
Brig.Schema.V93_AddScimPendingUserEmail
+ Brig.Schema.V94_ReduceUserGCGracePeriod
Brig.Team.API
Brig.Team.Template
Brig.Template
diff --git a/services/brig/src/Brig/CanonicalInterpreter.hs b/services/brig/src/Brig/CanonicalInterpreter.hs
index 4414567c910..e6103cb4722 100644
--- a/services/brig/src/Brig/CanonicalInterpreter.hs
+++ b/services/brig/src/Brig/CanonicalInterpreter.hs
@@ -431,7 +431,7 @@ runBrigToIO e (AppT ma) = do
case e.postgresMigration.user of
CassandraStorage -> interpretUserStoreCassandra e.casClient
PostgresqlStorage -> interpretUserStorePostgres
- MigrationToPostgresql -> error "Migration not implemented for user"
+ MigrationToPostgresql -> interpretUserStoreToCassandraAndPostgres e.casClient
( either throwM pure
<=< ( runFinal
diff --git a/services/brig/src/Brig/Index/Eval.hs b/services/brig/src/Brig/Index/Eval.hs
index 695007e195e..827d2028309 100644
--- a/services/brig/src/Brig/Index/Eval.hs
+++ b/services/brig/src/Brig/Index/Eval.hs
@@ -45,8 +45,11 @@ import Hasql.Pool.Extended qualified as Hasql
import Imports
import Network.HTTP.Client (Manager)
import Polysemy
+import Polysemy.Async (Async, asyncToIOFinal)
+import Polysemy.Conc (Race, interpretRace)
import Polysemy.Error
import Polysemy.Input
+import Polysemy.Resource (Resource, runResource)
import Polysemy.TinyLog (TinyLog)
import System.Logger qualified as Log
import System.Logger.Class (Logger)
@@ -60,6 +63,7 @@ import Wire.IndexedUserStore.Bulk.ElasticSearch qualified as IndexedUserStoreBul
import Wire.IndexedUserStore.ElasticSearch
import Wire.IndexedUserStore.MigrationStore (IndexedUserMigrationStore)
import Wire.IndexedUserStore.MigrationStore.ElasticSearch
+import Wire.MigrationLock
import Wire.ParseException
import Wire.PostgresMigrationOpts
import Wire.Rpc
@@ -83,6 +87,7 @@ type BrigIndexEffectStack =
Error IndexedUserStoreError,
IndexedUserMigrationStore,
Error MigrationException,
+ Error MigrationLockError,
GalleyAPIAccess,
Error ParseException,
Rpc,
@@ -92,6 +97,9 @@ type BrigIndexEffectStack =
Error UsageError,
Error TeamCollaboratorsError,
Error ClientError,
+ Resource,
+ Race,
+ Async,
Embed IO,
Final IO
]
@@ -132,10 +140,13 @@ runSem :: SemDeps -> UserStorageLocation -> Endpoint -> Logger -> Sem BrigIndexE
runSem (mgr, casClient, pgPool, bhEnv, indexedUserStoreConfig, reqId, migrationIndexName) userStorage galleyEndpoint logger action = do
let userStoreInterpreter = case userStorage.userStorageLocation of
CassandraStorage -> interpretUserStoreCassandra casClient
- MigrationToPostgresql -> error "Migration not implemented for user"
+ MigrationToPostgresql -> interpretUserStoreToCassandraAndPostgres casClient
PostgresqlStorage -> interpretUserStorePostgres
runFinal
. embedToFinal
+ . asyncToIOFinal
+ . interpretRace
+ . runResource
. throwErrorToIOFinal @ClientError
. throwErrorToIOFinal @TeamCollaboratorsError
. throwPostgresUsageErrorToIOFinal
@@ -145,6 +156,7 @@ runSem (mgr, casClient, pgPool, bhEnv, indexedUserStoreConfig, reqId, migrationI
. runRpcWithHttp mgr reqId
. throwErrorToIOFinal @ParseException
. interpretGalleyAPIAccessToRpc mempty galleyEndpoint
+ . throwErrorToIOFinal @MigrationLockError
. throwErrorToIOFinal @MigrationException
. interpretIndexedUserMigrationStoreES bhEnv migrationIndexName
. throwErrorToIOFinal @IndexedUserStoreError
diff --git a/services/brig/src/Brig/Run.hs b/services/brig/src/Brig/Run.hs
index 7cc07fc7f2d..fd170bdbaff 100644
--- a/services/brig/src/Brig/Run.hs
+++ b/services/brig/src/Brig/Run.hs
@@ -101,7 +101,7 @@ run opts = withTracer \tracer -> do
authMetrics <- Async.async (runBrigToIO e collectAuthMetrics)
pendingActivationCleanupAsync <- Async.async (runBrigToIO e pendingActivationCleanup)
- inSpan tracer "brig" defaultSpanArguments {kind = Otel.Server} (runSettingsWithShutdown s app Nothing) `finally` do
+ inSpan tracer "brig" defaultSpanArguments {kind = Otel.Server} (runSettingsWithCleanup (flush e.appLogger) s app Nothing) `finally` do
Async.cancelMany $
[internalEventListener, pendingActivationCleanupAsync, authMetrics]
<> catMaybes [emailListener, sftDiscovery]
diff --git a/services/brig/src/Brig/Schema/Run.hs b/services/brig/src/Brig/Schema/Run.hs
index 560cf64f2e7..ca29a34b79d 100644
--- a/services/brig/src/Brig/Schema/Run.hs
+++ b/services/brig/src/Brig/Schema/Run.hs
@@ -68,6 +68,7 @@ import Brig.Schema.V90_DomainRegistrationTeamIndex qualified as V90_DomainRegist
import Brig.Schema.V91_UpdateDomainRegistrationSchema_AddWebappUrl qualified as V91_UpdateDomainRegistrationSchema_AddWebappUrl
import Brig.Schema.V92_AddUserType qualified as V92_AddUserType
import Brig.Schema.V93_AddScimPendingUserEmail qualified as V93_AddScimPendingUserEmail
+import Brig.Schema.V94_ReduceUserGCGracePeriod qualified as V94_ReduceUserGCGracePeriod
import Cassandra.MigrateSchema (migrateSchema)
import Cassandra.Schema
import Control.Exception (finally)
@@ -142,7 +143,8 @@ migrations =
V90_DomainRegistrationTeamIndex.migration,
V91_UpdateDomainRegistrationSchema_AddWebappUrl.migration,
V92_AddUserType.migration,
- V93_AddScimPendingUserEmail.migration
+ V93_AddScimPendingUserEmail.migration,
+ V94_ReduceUserGCGracePeriod.migration
-- FUTUREWORK: undo V41 (searchable flag); we stopped using it in
-- https://github.com/wireapp/wire-server/pull/964
]
diff --git a/services/brig/src/Brig/Schema/V94_ReduceUserGCGracePeriod.hs b/services/brig/src/Brig/Schema/V94_ReduceUserGCGracePeriod.hs
new file mode 100644
index 00000000000..3761b2f7f0a
--- /dev/null
+++ b/services/brig/src/Brig/Schema/V94_ReduceUserGCGracePeriod.hs
@@ -0,0 +1,44 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+-- This file is part of the Wire Server implementation.
+--
+-- Copyright (C) 2026 Wire Swiss GmbH
+--
+-- 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 .
+module Brig.Schema.V94_ReduceUserGCGracePeriod
+ ( migration,
+ )
+where
+
+import Cassandra.Schema
+import Imports
+import Text.RawString.QQ
+
+migration :: Migration
+migration =
+ Migration 94 "reduce user gc_grace_period" $ do
+ schema'
+ [r| ALTER TABLE user WITH gc_grace_seconds = 86400 |]
+
+ schema'
+ [r| ALTER TABLE user_handle WITH gc_grace_seconds = 86400 |]
+
+ schema'
+ [r| ALTER TABLE rich_info WITH gc_grace_seconds = 86400 |]
+
+ schema'
+ [r| ALTER TABLE service_user WITH gc_grace_seconds = 86400 |]
+
+ schema'
+ [r| ALTER TABLE service_team WITH gc_grace_seconds = 86400 |]