From bd9559c6a5e778f64c1f59748e623f2fff7a6513 Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Mon, 3 Aug 2026 20:22:47 +0200 Subject: [PATCH] WPB-22965: migrate blacklist to PostGreSQL Migrate the brig blacklist (BlockListStore) from Cassandra to PostgreSQL using the dual-write + background-worker pattern, following the shipped domainRegistration migration as the reference. Adds the Postgres/DualWrite/Migration interpreters, the blockList StorageLocation config field, background-worker wiring, the SQL migration plus regenerated postgres-schema.sql, chart/config/gotmpl/docs updates, and an integration test. --- changelog.d/5-internal/WPB-22965 | 1 + .../background-worker/configmap.yaml | 1 + charts/wire-server/values.yaml | 5 + .../src/developer/reference/config-options.md | 10 ++ hack/helm_vars/common.yaml.gotmpl | 1 + hack/helm_vars/wire-server/values.yaml.gotmpl | 1 + integration/integration.cabal | 1 + integration/test/API/BrigInternal.hs | 15 ++ integration/test/Test/Migration/BlockList.hs | 68 ++++++++ .../20260803164250-blacklist.sql | 3 + .../src/Wire/BlockListStore/Cassandra.hs | 4 + .../src/Wire/BlockListStore/DualWrite.hs | 45 +++++ .../src/Wire/BlockListStore/Migration.hs | 155 ++++++++++++++++++ .../src/Wire/BlockListStore/Postgres.hs | 74 +++++++++ .../src/Wire/PostgresMigrationOpts.hs | 4 +- libs/wire-subsystems/wire-subsystems.cabal | 3 + postgres-schema.sql | 18 ++ .../background-worker.integration.yaml | 2 + .../src/Wire/BackgroundWorker.hs | 10 +- .../src/Wire/BackgroundWorker/Options.hs | 1 + .../src/Wire/PostgresMigrations.hs | 19 +++ .../Wire/BackendNotificationPusherSpec.hs | 6 +- .../background-worker/test/Test/Wire/Util.hs | 3 +- services/brig/brig.integration.yaml | 1 + .../brig/src/Brig/CanonicalInterpreter.hs | 11 +- services/galley/galley.integration.yaml | 1 + 26 files changed, 456 insertions(+), 7 deletions(-) create mode 100644 changelog.d/5-internal/WPB-22965 create mode 100644 integration/test/Test/Migration/BlockList.hs create mode 100644 libs/wire-subsystems/postgres-migrations/20260803164250-blacklist.sql create mode 100644 libs/wire-subsystems/src/Wire/BlockListStore/DualWrite.hs create mode 100644 libs/wire-subsystems/src/Wire/BlockListStore/Migration.hs create mode 100644 libs/wire-subsystems/src/Wire/BlockListStore/Postgres.hs diff --git a/changelog.d/5-internal/WPB-22965 b/changelog.d/5-internal/WPB-22965 new file mode 100644 index 00000000000..46a3360e74e --- /dev/null +++ b/changelog.d/5-internal/WPB-22965 @@ -0,0 +1 @@ +brig: migrate blacklist (BlockListStore) from Cassandra to PostgreSQL diff --git a/charts/wire-server/templates/background-worker/configmap.yaml b/charts/wire-server/templates/background-worker/configmap.yaml index d4fe2a63202..c34ca036ed2 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 }} + migrateBlockList: {{ .migrateBlockList }} migrationOptions: {{ toYaml .migrationOptions | indent 6 }} diff --git a/charts/wire-server/values.yaml b/charts/wire-server/values.yaml index ae71d850795..e6660692f78 100644 --- a/charts/wire-server/values.yaml +++ b/charts/wire-server/values.yaml @@ -90,6 +90,7 @@ galley: teamFeatures: cassandra domainRegistration: cassandra user: cassandra + blockList: cassandra settings: httpPoolSize: 128 maxTeamSize: 10000 @@ -1017,6 +1018,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 blacklist data. + # It's important to set `settings.postgresMigration.blockList` to `migration-to-postgresql` + # before starting the migration. + migrateBlockList: false backendNotificationPusher: pushBackoffMinWait: 10000 # in microseconds, so 10ms diff --git a/docs/src/developer/reference/config-options.md b/docs/src/developer/reference/config-options.md index 066637dbb07..d27462e0fb4 100644 --- a/docs/src/developer/reference/config-options.md +++ b/docs/src/developer/reference/config-options.md @@ -2106,12 +2106,14 @@ galley: teamFeatures: postgresql domainRegistration: postgresql user: postgresql + blockList: postgresql background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false + migrateBlockList: false ``` #### Migration for existing installations @@ -2142,6 +2144,7 @@ The current settings and their background-worker flags are: - `conversationCodes` -> `migrateConversationCodes` - `teamFeatures` -> `migrateTeamFeatures` - `domainRegistration` -> `migrateDomainRegistration` +- `blockList` -> `migrateBlockList` **Migration pattern per migration setting** @@ -2161,12 +2164,14 @@ The current settings and their background-worker flags are: conversationCodes: migration-to-postgresql teamFeatures: migration-to-postgresql domainRegistration: cassandra + blockList: cassandra background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false + migrateBlockList: false ``` This change should restart the affected pods, and new writes will follow the @@ -2181,6 +2186,7 @@ The current settings and their background-worker flags are: migrateConversationCodes: true migrateTeamFeatures: true migrateDomainRegistration: true + migrateBlockList: true ``` During migration, Cassandra rows are not deleted. Writes and migration share @@ -2196,6 +2202,7 @@ 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` + - `blockList`: `wire_block_list_migration_finished` 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 @@ -2209,12 +2216,14 @@ The current settings and their background-worker flags are: conversationCodes: postgresql teamFeatures: postgresql domainRegistration: cassandra + blockList: cassandra background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false + migrateBlockList: false ``` **How to run migrations independently or in batches** @@ -2303,6 +2312,7 @@ migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false +migrateBlockList: false # migration settings migrationOptions: diff --git a/hack/helm_vars/common.yaml.gotmpl b/hack/helm_vars/common.yaml.gotmpl index 2276355e2a9..4d36ffac030 100644 --- a/hack/helm_vars/common.yaml.gotmpl +++ b/hack/helm_vars/common.yaml.gotmpl @@ -19,6 +19,7 @@ conversationCodesStore: {{ $preferredStore }} teamFeaturesStore: {{ $preferredStore }} domainRegistration: {{ $preferredStore }} userStore: {{ $preferredStore }} +blockListStore: {{ $preferredStore }} {{- if (eq (env "UPLOAD_XML_S3_BASE_URL") "") }} uploadXml: {} diff --git a/hack/helm_vars/wire-server/values.yaml.gotmpl b/hack/helm_vars/wire-server/values.yaml.gotmpl index bd6cdadf2a8..a30f413d901 100644 --- a/hack/helm_vars/wire-server/values.yaml.gotmpl +++ b/hack/helm_vars/wire-server/values.yaml.gotmpl @@ -306,6 +306,7 @@ galley: teamFeatures: {{ .Values.teamFeaturesStore }} domainRegistration: {{ .Values.domainRegistration }} user: {{ .Values.userStore }} + blockList: {{ .Values.blockListStore }} settings: maxConvAndTeamSize: 16 maxTeamSize: 32 diff --git a/integration/integration.cabal b/integration/integration.cabal index c36b9c4e730..b3621b6234f 100644 --- a/integration/integration.cabal +++ b/integration/integration.cabal @@ -177,6 +177,7 @@ library Test.Login Test.Meetings Test.MessageTimer + Test.Migration.BlockList Test.Migration.Conversation Test.Migration.ConversationCodes Test.Migration.DomainRegistration diff --git a/integration/test/API/BrigInternal.hs b/integration/test/API/BrigInternal.hs index 4a0051e8907..9cb4e0d8d24 100644 --- a/integration/test/API/BrigInternal.hs +++ b/integration/test/API/BrigInternal.hs @@ -343,6 +343,21 @@ getPasswordResetCode domain email = do req <- baseRequest domain Brig Unversioned "i/users/password-reset-code" submit "GET" $ req & addQueryParams [("email", email)] +addBlacklist :: (HasCallStack, MakesValue domain) => domain -> String -> App Response +addBlacklist domain email = do + req <- baseRequest domain Brig Unversioned "i/users/blacklist" + submit "POST" $ req & addQueryParams [("email", email)] + +deleteBlacklist :: (HasCallStack, MakesValue domain) => domain -> String -> App Response +deleteBlacklist domain email = do + req <- baseRequest domain Brig Unversioned "i/users/blacklist" + submit "DELETE" $ req & addQueryParams [("email", email)] + +checkBlacklist :: (HasCallStack, MakesValue domain) => domain -> String -> App Response +checkBlacklist domain email = do + req <- baseRequest domain Brig Unversioned "i/users/blacklist" + submit "GET" $ req & addQueryParams [("email", email)] + data PutSSOId = PutSSOId { scimExternalId :: Maybe String, subject :: Maybe String, diff --git a/integration/test/Test/Migration/BlockList.hs b/integration/test/Test/Migration/BlockList.hs new file mode 100644 index 00000000000..824c9763853 --- /dev/null +++ b/integration/test/Test/Migration/BlockList.hs @@ -0,0 +1,68 @@ +-- 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 Test.Migration.BlockList (testBlockListMigration) where + +import qualified API.BrigInternal as BrigInternal +import API.Common +import Control.Monad.Codensity +import Control.Monad.Reader +import Test.Migration.Util (waitForMigration) +import Testlib.Prelude +import Testlib.ResourcePool + +-- | Migrate the 'blacklist' store (brig) from Cassandra to PostgreSQL. +-- +-- The blacklist holds email keys with no read-back payload, so the migration is +-- a straight copy: a key blacklisted in Cassandra must survive the cutover and +-- remain deletable once PostgreSQL is the sole source of truth. +testBlockListMigration :: (HasCallStack) => App () +testBlockListMigration = do + resourcePool <- asks (.resourcePool) + email <- randomEmail + runCodensity (acquireResources 1 resourcePool) $ \[backend] -> do + let domain = backend.berDomain + + -- Cassandra: blacklist an email key and confirm it is reported as such. + runCodensity (startDynamicBackend backend (conf "cassandra" False)) $ \_ -> do + assertSuccess =<< BrigInternal.addBlacklist domain email + assertStatus 200 =<< BrigInternal.checkBlacklist domain email + + -- migration-to-postgresql with the worker running: backfill the existing key + -- and confirm it is still blacklisted once the migration is finished. + runCodensity (startDynamicBackend backend (conf "migration-to-postgresql" True)) $ \_ -> do + waitForMigration domain counterName + assertStatus 200 =<< BrigInternal.checkBlacklist domain email + + -- PostgreSQL only: the migrated key must persist, and deleting it must + -- remove it. + runCodensity (startDynamicBackend backend (conf "postgresql" False)) $ \_ -> do + assertStatus 200 =<< BrigInternal.checkBlacklist domain email + assertSuccess =<< BrigInternal.deleteBlacklist domain email + assertStatus 404 =<< BrigInternal.checkBlacklist domain email + where + conf :: String -> Bool -> ServiceOverrides + conf db runMigration = + def + { brigCfg = setField "postgresMigration.blockList" db, + backgroundWorkerCfg = + setField "postgresMigration.blockList" db + >=> setField "migrateBlockList" runMigration + } + + counterName :: String + counterName = "^wire_block_list_migration_finished" diff --git a/libs/wire-subsystems/postgres-migrations/20260803164250-blacklist.sql b/libs/wire-subsystems/postgres-migrations/20260803164250-blacklist.sql new file mode 100644 index 00000000000..91c8fc3a144 --- /dev/null +++ b/libs/wire-subsystems/postgres-migrations/20260803164250-blacklist.sql @@ -0,0 +1,3 @@ +CREATE TABLE IF NOT EXISTS blacklist ( + key text PRIMARY KEY +); diff --git a/libs/wire-subsystems/src/Wire/BlockListStore/Cassandra.hs b/libs/wire-subsystems/src/Wire/BlockListStore/Cassandra.hs index 0bf4f46d3ac..7cfbda00b31 100644 --- a/libs/wire-subsystems/src/Wire/BlockListStore/Cassandra.hs +++ b/libs/wire-subsystems/src/Wire/BlockListStore/Cassandra.hs @@ -17,6 +17,7 @@ module Wire.BlockListStore.Cassandra ( interpretBlockListStoreToCassandra, + selectAllBlacklist, ) where @@ -60,3 +61,6 @@ keySelect = "SELECT key FROM blacklist WHERE key = ?" keyDelete :: PrepQuery W (Identity Text) () keyDelete = "DELETE FROM blacklist WHERE key = ?" + +selectAllBlacklist :: PrepQuery R () (Identity Text) +selectAllBlacklist = "SELECT key FROM blacklist" diff --git a/libs/wire-subsystems/src/Wire/BlockListStore/DualWrite.hs b/libs/wire-subsystems/src/Wire/BlockListStore/DualWrite.hs new file mode 100644 index 00000000000..b95371b7c34 --- /dev/null +++ b/libs/wire-subsystems/src/Wire/BlockListStore/DualWrite.hs @@ -0,0 +1,45 @@ +-- 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.BlockListStore.DualWrite + ( interpretBlockListStoreToCassandraAndPostgres, + ) +where + +import Cassandra (ClientState) +import Imports +import Polysemy +import Wire.BlockListStore +import Wire.BlockListStore qualified as BlockListStore +import Wire.BlockListStore.Cassandra qualified as Cassandra +import Wire.BlockListStore.Postgres qualified as Postgres +import Wire.Postgres (PGConstraints) + +-- | Cassandra is the source of truth during migration; writes are mirrored to Postgres. +interpretBlockListStoreToCassandraAndPostgres :: + (PGConstraints r) => + ClientState -> + InterpreterFor BlockListStore r +interpretBlockListStoreToCassandraAndPostgres cs = interpret $ \case + Insert key -> do + Cassandra.interpretBlockListStoreToCassandra cs $ BlockListStore.insert key + Postgres.interpretBlockListStoreToPostgres $ BlockListStore.insert key + Exists key -> + Cassandra.interpretBlockListStoreToCassandra cs $ BlockListStore.exists key + Delete key -> do + Cassandra.interpretBlockListStoreToCassandra cs $ BlockListStore.delete key + Postgres.interpretBlockListStoreToPostgres $ BlockListStore.delete key diff --git a/libs/wire-subsystems/src/Wire/BlockListStore/Migration.hs b/libs/wire-subsystems/src/Wire/BlockListStore/Migration.hs new file mode 100644 index 00000000000..3c1b23aec4f --- /dev/null +++ b/libs/wire-subsystems/src/Wire/BlockListStore/Migration.hs @@ -0,0 +1,155 @@ +-- 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.BlockListStore.Migration (migrateBlacklistLoop) where + +import Cassandra hiding (Value) +import Data.ByteString.Conversion (toByteString') +import Data.Conduit +import Data.Conduit.List qualified as C +import Data.IORef qualified as IORef +import Data.Time +import Hasql.Pool.Extended qualified as Hasql +import Imports +import Polysemy +import Polysemy.Async +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 +import UnliftIO qualified +import Wire.BlockListStore.Cassandra qualified as Cql +import Wire.BlockListStore.Postgres qualified as Postgres +import Wire.Migration +import Wire.Postgres +import Wire.Sem.Logger (mapLogger) +import Wire.Sem.Logger.TinyLog (loggerToTinyLog) + +type EffectStack = + [ State Int, + Input ClientState, + Input Hasql.Pool, + Resource, + Async, + Race, + TinyLog, + Embed IO, + Final IO + ] + +migrateBlacklistLoop :: + MigrationOptions -> + ClientState -> + Hasql.Pool -> + Log.Logger -> + Prometheus.Counter -> + Prometheus.Counter -> + Prometheus.Counter -> + Prometheus.Vector Text Prometheus.Histogram -> + IO () +migrateBlacklistLoop migOpts cassClient pgPool logger migCounter migFinished migFailed migDuration = + migrationLoop + logger + "blacklist" + migFinished + migFailed + (interpreter cassClient pgPool logger "blacklist") + (migrateAllBlacklist migOpts migCounter migDuration) + +interpreter :: ClientState -> Hasql.Pool -> Log.Logger -> ByteString -> Sem EffectStack a -> IO (Int, a) +interpreter cassClient pgPool logger name = + runFinal + . embedToFinal + . loggerToTinyLog logger + . mapLogger (Log.field "migration" (Log.val name) .) + . raiseUnder + . interpretRace + . asyncToIOFinal + . resourceToIOFinal + . runInputConst pgPool + . runInputConst cassClient + . runState 0 + +migrateAllBlacklist :: + ( Member (Input Hasql.Pool) r, + Member (Embed IO) r, + Member (Input ClientState) r, + Member TinyLog r, + Member (State Int) r, + Member Resource r, + Member Race r + ) => + MigrationOptions -> + Prometheus.Counter -> + Prometheus.Vector Text Prometheus.Histogram -> + ConduitM () Void (Sem r) () +migrateAllBlacklist migOpts migCounter migDuration = do + lift $ info $ Log.msg (Log.val "migrateAllBlacklist") + withCount + (paginateSem Cql.selectAllBlacklist (paramsP LocalQuorum () migOpts.pageSize) x5) + .| logRetrievedPage migOpts.pageSize id + .| C.mapM_ + ( traverse_ + ( \row@(Identity key) -> + handleErrors (toByteString' key) (migrateBlacklistRow migOpts migCounter migDuration row) + ) + ) + +migrateBlacklistRow :: + ( PGConstraints r, + Member TinyLog r, + Member Resource r, + Member Race r + ) => + MigrationOptions -> + Prometheus.Counter -> + Prometheus.Vector Text Prometheus.Histogram -> + Identity Text -> + Sem r () +migrateBlacklistRow migOpts migCounter migDuration (Identity key) = do + outcomeRef <- liftIO $ IORef.newIORef @Text "error" + bracket + (liftIO getCurrentTime) + (observeDuration migDuration outcomeRef) + ( const $ do + timeoutResult <- Conc.timeout (migOpts.timeout <$ handleTimeout) migOpts.timeout $ Postgres.insertKey key + case timeoutResult of + Left timedOutAfter -> do + markOutcome outcomeRef "timeout" + liftIO . UnliftIO.throwIO $ MigrationTimedOut key timedOutAfter + Right () -> do + markOutcome outcomeRef "success" + liftIO $ Prometheus.incCounter migCounter + ) + where + handleTimeout = + err $ + Log.msg (Log.val "blacklist migration timed out") + . Log.field "key" (show key) + . Log.field "timeout" (show migOpts.timeout) + + markOutcome ref outcome = liftIO $ IORef.writeIORef ref outcome + + observeDuration metric outcomeRef start = do + outcome <- liftIO $ IORef.readIORef outcomeRef + end <- liftIO getCurrentTime + liftIO $ Prometheus.withLabel metric outcome (`Prometheus.observe` realToFrac (diffUTCTime end start)) diff --git a/libs/wire-subsystems/src/Wire/BlockListStore/Postgres.hs b/libs/wire-subsystems/src/Wire/BlockListStore/Postgres.hs new file mode 100644 index 00000000000..5a7d1d8f283 --- /dev/null +++ b/libs/wire-subsystems/src/Wire/BlockListStore/Postgres.hs @@ -0,0 +1,74 @@ +-- 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.BlockListStore.Postgres + ( interpretBlockListStoreToPostgres, + insertKey, + ) +where + +import Hasql.Statement qualified as Hasql +import Hasql.TH +import Imports +import Polysemy +import Wire.BlockListStore (BlockListStore (..)) +import Wire.Postgres +import Wire.UserKeyStore (EmailKey, emailKeyUniq) + +interpretBlockListStoreToPostgres :: + (PGConstraints r) => + InterpreterFor BlockListStore r +interpretBlockListStoreToPostgres = interpret $ \case + Insert key -> insertImpl key + Exists key -> existsImpl key + Delete key -> deleteImpl key + +insertImpl :: (PGConstraints r) => EmailKey -> Sem r () +insertImpl = insertKey . emailKeyUniq + +insertKey :: (PGConstraints r) => Text -> Sem r () +insertKey key = + runStatement key insertStatement + where + insertStatement :: Hasql.Statement Text () + insertStatement = + [resultlessStatement|INSERT INTO blacklist (key) + VALUES ($1 :: text) + ON CONFLICT DO NOTHING + |] + +existsImpl :: (PGConstraints r) => EmailKey -> Sem r Bool +existsImpl key = + runStatement (emailKeyUniq key) existsStatement + where + existsStatement :: Hasql.Statement Text Bool + existsStatement = + [singletonStatement|SELECT EXISTS ( + SELECT 1 + FROM blacklist + WHERE key = ($1 :: text) + ) :: bool|] + +deleteImpl :: (PGConstraints r) => EmailKey -> Sem r () +deleteImpl key = + runStatement (emailKeyUniq key) deleteStatement + where + deleteStatement :: Hasql.Statement Text () + deleteStatement = + [resultlessStatement|DELETE FROM blacklist + WHERE key = ($1 :: text) + |] diff --git a/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs b/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs index 327862f7cd5..d5e350e608f 100644 --- a/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs +++ b/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs @@ -56,7 +56,8 @@ data PostgresMigrationOpts = PostgresMigrationOpts conversationCodes :: StorageLocation, teamFeatures :: StorageLocation, domainRegistration :: StorageLocation, - user :: StorageLocation + user :: StorageLocation, + blockList :: StorageLocation } deriving (Show) @@ -68,3 +69,4 @@ instance FromJSON PostgresMigrationOpts where <*> o .: "teamFeatures" <*> o .: "domainRegistration" <*> o .: "user" + <*> o .: "blockList" diff --git a/libs/wire-subsystems/wire-subsystems.cabal b/libs/wire-subsystems/wire-subsystems.cabal index 943f8fa0294..dc67cf877c6 100644 --- a/libs/wire-subsystems/wire-subsystems.cabal +++ b/libs/wire-subsystems/wire-subsystems.cabal @@ -241,6 +241,9 @@ library Wire.BackgroundJobsRunner.Interpreter Wire.BlockListStore Wire.BlockListStore.Cassandra + Wire.BlockListStore.DualWrite + Wire.BlockListStore.Migration + Wire.BlockListStore.Postgres Wire.BoundedQueue Wire.BoundedQueue.STM Wire.BrigAPIAccess diff --git a/postgres-schema.sql b/postgres-schema.sql index 948f1d62594..ab6f599bed4 100644 --- a/postgres-schema.sql +++ b/postgres-schema.sql @@ -104,6 +104,17 @@ CREATE TABLE public.asset ( ALTER TABLE public.asset OWNER TO "wire-server"; +-- +-- Name: blacklist; Type: TABLE; Schema: public; Owner: wire-server +-- + +CREATE TABLE public.blacklist ( + key text NOT NULL +); + + +ALTER TABLE public.blacklist OWNER TO "wire-server"; + -- -- Name: bot_conv; Type: TABLE; Schema: public; Owner: wire-server -- @@ -488,6 +499,13 @@ ALTER TABLE public.wire_user OWNER TO "wire-server"; ALTER TABLE ONLY public.apps ADD CONSTRAINT apps_pkey PRIMARY KEY (user_id); +-- +-- Name: blacklist blacklist_pkey; Type: CONSTRAINT; Schema: public; Owner: wire-server +-- + +ALTER TABLE ONLY public.blacklist + ADD CONSTRAINT blacklist_pkey PRIMARY KEY (key); + -- -- Name: bot_conv bot_conv_pkey; 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..25eceaadbc9 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 +migrateBlockList: false # Background jobs consumer configuration for integration backgroundJobs: @@ -92,3 +93,4 @@ postgresMigration: teamFeatures: postgresql domainRegistration: postgresql user: postgresql + blockList: postgresql diff --git a/services/background-worker/src/Wire/BackgroundWorker.hs b/services/background-worker/src/Wire/BackgroundWorker.hs index b57ba12df40..3f5cb84ecee 100644 --- a/services/background-worker/src/Wire/BackgroundWorker.hs +++ b/services/background-worker/src/Wire/BackgroundWorker.hs @@ -78,6 +78,13 @@ run opts galleyOpts = do withNamedLogger "migrate-domain-registration" $ Migrations.domainRegistration opts.migrationOptions else pure $ pure () + cleanupBlockListMigration <- + if opts.migrateBlockList + then + runAppT env $ + withNamedLogger "migrate-block-list" $ + Migrations.blockList opts.migrationOptions + else pure $ pure () cleanupJobs <- runAppT env $ withNamedLogger "background-job-consumer" $ @@ -89,13 +96,14 @@ run opts galleyOpts = do let cleanup = void $ runConcurrently $ - (,,,,,,,) + (,,,,,,,,) <$> Concurrently cleanupDeadUserNotifWatcher <*> Concurrently cleanupBackendNotifPusher <*> Concurrently cleanupConvMigration <*> Concurrently cleanUpConvCodesMigration <*> Concurrently cleanupTeamFeaturesMigration <*> Concurrently cleanupDomainRegistrationMigration + <*> Concurrently cleanupBlockListMigration <*> 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..16f2360fb05 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, + migrateBlockList :: !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..bf0d1bf8bd2 100644 --- a/services/background-worker/src/Wire/PostgresMigrations.hs +++ b/services/background-worker/src/Wire/PostgresMigrations.hs @@ -23,6 +23,7 @@ import System.Logger qualified as Log import UnliftIO import Wire.BackgroundWorker.Env import Wire.BackgroundWorker.Util +import Wire.BlockListStore.Migration import Wire.CodeStore.Migration import Wire.ConversationStore.Migration import Wire.DomainRegistrationStore.Migration @@ -107,3 +108,21 @@ domainRegistration migOpts = do pure $ do Log.info logger $ Log.msg (Log.val "cancelling domain registration migration") cancel migrationLoop + +blockList :: MigrationOptions -> AppT IO CleanupAction +blockList migOpts = do + cassClient <- asks (.cassandraBrig) + pgPool <- asks (.hasqlPool) + logger <- asks (.logger) + Log.info logger $ Log.msg (Log.val "starting blacklist migration") + count <- register $ counter $ Prometheus.Info "wire_block_list_migrated_to_pg" "Number of blacklist keys migrated to Postgresql" + finished <- register $ counter $ Prometheus.Info "wire_block_list_migration_finished" "Whether the blacklist migration to Postgresql is finished successfully" + failed <- register $ counter $ Prometheus.Info "wire_block_list_migration_failed" "Whether the blacklist migration to Postgresql has failed" + duration <- register $ vector "outcome" $ histogram (Prometheus.Info "wire_block_list_migration_duration_seconds" "Duration of blacklist migration attempts") defaultBuckets + + migrationLoop <- async . lift $ migrateBlacklistLoop migOpts cassClient pgPool logger count finished failed duration + + Log.info logger $ Log.msg (Log.val "started blacklist migration") + pure $ do + Log.info logger $ Log.msg (Log.val "cancelling blacklist migration") + cancel migrationLoop diff --git a/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs b/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs index 7222120d93a..dcfc0e8dc80 100644 --- a/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs +++ b/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs @@ -371,7 +371,8 @@ spec = do conversationCodes = CassandraStorage, teamFeatures = CassandraStorage, domainRegistration = CassandraStorage, - user = CassandraStorage + user = CassandraStorage, + blockList = CassandraStorage } gundeckEndpoint = undefined brigEndpoint = undefined @@ -435,7 +436,8 @@ spec = do conversationCodes = CassandraStorage, teamFeatures = CassandraStorage, domainRegistration = CassandraStorage, - user = CassandraStorage + user = CassandraStorage, + blockList = CassandraStorage } gundeckEndpoint = undefined brigEndpoint = undefined diff --git a/services/background-worker/test/Test/Wire/Util.hs b/services/background-worker/test/Test/Wire/Util.hs index 5d89532bfec..74355c339fc 100644 --- a/services/background-worker/test/Test/Wire/Util.hs +++ b/services/background-worker/test/Test/Wire/Util.hs @@ -50,7 +50,8 @@ testEnv = do conversationCodes = CassandraStorage, teamFeatures = CassandraStorage, domainRegistration = CassandraStorage, - user = CassandraStorage + user = CassandraStorage, + blockList = CassandraStorage } statuses <- newIORef mempty backendNotificationMetrics <- mkBackendNotificationMetrics diff --git a/services/brig/brig.integration.yaml b/services/brig/brig.integration.yaml index 8be11f028bd..d8a8a9957e2 100644 --- a/services/brig/brig.integration.yaml +++ b/services/brig/brig.integration.yaml @@ -176,6 +176,7 @@ postgresMigration: teamFeatures: postgresql domainRegistration: postgresql user: postgresql + blockList: postgresql optSettings: setActivationTimeout: 4 diff --git a/services/brig/src/Brig/CanonicalInterpreter.hs b/services/brig/src/Brig/CanonicalInterpreter.hs index a39da9d12a3..c9d521d253c 100644 --- a/services/brig/src/Brig/CanonicalInterpreter.hs +++ b/services/brig/src/Brig/CanonicalInterpreter.hs @@ -69,6 +69,8 @@ import Wire.BackgroundJobsPublisher (BackgroundJobPublisher) import Wire.BackgroundJobsPublisher.RabbitMQ (interpretBackgroundJobPublisherRabbitMQ) import Wire.BlockListStore import Wire.BlockListStore.Cassandra +import Wire.BlockListStore.DualWrite +import Wire.BlockListStore.Postgres import Wire.ClientStore (ClientStore) import Wire.ClientStore.Cassandra import Wire.ClientStore.DynamoDB (OptimisticLockEnv (..)) @@ -218,6 +220,7 @@ type BrigLowerLevelEffects = UserGroupStore, DomainRegistrationStore, DomainVerificationChallengeStore, + BlockListStore, Error AppSubsystemError, Error TeamCollaboratorsError, Error UsageError, @@ -261,7 +264,6 @@ type BrigLowerLevelEffects = Jwk, PublicKeyBundle, JwtTools, - BlockListStore, UserPendingActivationStore InternalPaging, Now, Delay, @@ -409,6 +411,11 @@ runBrigToIO e (AppT ma) = do PostgresqlStorage -> interpretDomainRegistrationStoreToPostgres MigrationToPostgresql -> interpretDomainRegistrationStoreToCassandraAndPostgres e.casClient + blockListStore = case e.postgresMigration.blockList of + CassandraStorage -> interpretBlockListStoreToCassandra e.casClient + PostgresqlStorage -> interpretBlockListStoreToPostgres + MigrationToPostgresql -> interpretBlockListStoreToCassandraAndPostgres e.casClient + domainVerificationChallengeStore = case e.postgresMigration.domainRegistration of CassandraStorage -> interpretDomainVerificationChallengeStoreToCassandra e.settings.challengeTTL PostgresqlStorage -> interpretDomainVerificationChallengeStoreToPostgres e.settings.challengeTTL @@ -445,7 +452,6 @@ runBrigToIO e (AppT ma) = do . runDelay . nowToIOAction e.currentTime . userPendingActivationStoreToCassandra - . interpretBlockListStoreToCassandra e.casClient . interpretJwtTools . interpretPublicKeyBundle . interpretJwk @@ -489,6 +495,7 @@ runBrigToIO e (AppT ma) = do . mapError postgresUsageErrorToHttpError . mapError teamCollaboratorsSubsystemErrorToHttpError . mapError appSubsystemErrorToHttpError + . blockListStore . domainVerificationChallengeStore . domainRegistrationStore . interpretUserGroupStoreToPostgres diff --git a/services/galley/galley.integration.yaml b/services/galley/galley.integration.yaml index 47980040c1e..4cf85df8082 100644 --- a/services/galley/galley.integration.yaml +++ b/services/galley/galley.integration.yaml @@ -266,3 +266,4 @@ postgresMigration: teamFeatures: postgresql domainRegistration: postgresql user: postgresql + blockList: postgresql