From d4a4cc4f7c03955d4826fcca1573cba38ae47f76 Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Tue, 4 Aug 2026 16:08:45 +0200 Subject: [PATCH] WPB-22961: migrate service (old bot) to PostGreSQL Migrate galley's ServiceStore (old bot provider/service connection data) from Cassandra to PostgreSQL using the dual-write + background-worker pattern: a Postgres interpreter, a DualWrite interpreter (reads from Cassandra, writes to both), and a copy-only backfill loop. Adds the service StorageLocation config field, a service table, marshall instances for Fingerprint Rsa and ServiceToken, and an integration test. --- changelog.d/5-internal/WPB-22961 | 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/Test/Migration/Service.hs | 76 +++++++++ .../wire-api/src/Wire/API/PostgresMarshall.hs | 6 + .../wire-api/src/Wire/API/Provider/Service.hs | 8 + .../Wire/API/Roundtrip/PostgresMarshall.hs | 7 +- .../20260804125818-service.sql | 9 + .../src/Wire/PostgresMigrationOpts.hs | 4 +- .../Wire/ServiceStore/Cassandra/Queries.hs | 27 +++ .../src/Wire/ServiceStore/DualWrite.hs | 49 ++++++ .../src/Wire/ServiceStore/Migration.hs | 159 ++++++++++++++++++ .../src/Wire/ServiceStore/Postgres.hs | 105 ++++++++++++ libs/wire-subsystems/wire-subsystems.cabal | 4 + postgres-schema.sql | 28 ++- .../background-worker.integration.yaml | 1 + .../src/Wire/BackgroundWorker.hs | 11 +- .../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 + services/galley/galley.integration.yaml | 1 + services/galley/src/Galley/App.hs | 9 +- 28 files changed, 544 insertions(+), 10 deletions(-) create mode 100644 changelog.d/5-internal/WPB-22961 create mode 100644 integration/test/Test/Migration/Service.hs create mode 100644 libs/wire-subsystems/postgres-migrations/20260804125818-service.sql create mode 100644 libs/wire-subsystems/src/Wire/ServiceStore/Cassandra/Queries.hs create mode 100644 libs/wire-subsystems/src/Wire/ServiceStore/DualWrite.hs create mode 100644 libs/wire-subsystems/src/Wire/ServiceStore/Migration.hs create mode 100644 libs/wire-subsystems/src/Wire/ServiceStore/Postgres.hs diff --git a/changelog.d/5-internal/WPB-22961 b/changelog.d/5-internal/WPB-22961 new file mode 100644 index 00000000000..fa1413ca230 --- /dev/null +++ b/changelog.d/5-internal/WPB-22961 @@ -0,0 +1 @@ +Migration of service (old bots) from cassandra to postgres diff --git a/charts/wire-server/templates/background-worker/configmap.yaml b/charts/wire-server/templates/background-worker/configmap.yaml index d4fe2a63202..c9355d6c149 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 }} + migrateService: {{ .migrateService }} migrationOptions: {{ toYaml .migrationOptions | indent 6 }} diff --git a/charts/wire-server/values.yaml b/charts/wire-server/values.yaml index ae71d850795..df75e858ad9 100644 --- a/charts/wire-server/values.yaml +++ b/charts/wire-server/values.yaml @@ -90,6 +90,7 @@ galley: teamFeatures: cassandra domainRegistration: cassandra user: cassandra + service: 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 services (old bots). + # It's important to set `settings.postgresMigration.service` to `migration-to-postgresql` + # before starting the migration. + migrateService: 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..54bf2588165 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 + service: postgresql background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false + migrateService: false ``` #### Migration for existing installations @@ -2142,6 +2144,7 @@ The current settings and their background-worker flags are: - `conversationCodes` -> `migrateConversationCodes` - `teamFeatures` -> `migrateTeamFeatures` - `domainRegistration` -> `migrateDomainRegistration` +- `service` -> `migrateService` **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 + service: cassandra background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false + migrateService: 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 + migrateService: 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` + - `service`: `wire_service_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 + service: postgresql background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false + migrateService: false ``` **How to run migrations independently or in batches** @@ -2303,6 +2312,7 @@ migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false +migrateService: false # migration settings migrationOptions: diff --git a/hack/helm_vars/common.yaml.gotmpl b/hack/helm_vars/common.yaml.gotmpl index 2276355e2a9..788ab28f467 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 }} +serviceStore: {{ $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..7b6d259cbf9 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 }} + service: {{ .Values.serviceStore }} settings: maxConvAndTeamSize: 16 maxTeamSize: 32 diff --git a/integration/integration.cabal b/integration/integration.cabal index c36b9c4e730..6501bc98ad6 100644 --- a/integration/integration.cabal +++ b/integration/integration.cabal @@ -180,6 +180,7 @@ library Test.Migration.Conversation Test.Migration.ConversationCodes Test.Migration.DomainRegistration + Test.Migration.Service Test.Migration.TeamFeatures Test.Migration.Util Test.MLS diff --git a/integration/test/Test/Migration/Service.hs b/integration/test/Test/Migration/Service.hs new file mode 100644 index 00000000000..3619aa739e9 --- /dev/null +++ b/integration/test/Test/Migration/Service.hs @@ -0,0 +1,76 @@ +-- 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.Service where + +import API.Brig +import API.Galley +import Control.Monad.Codensity +import Control.Monad.Reader +import Data.String.Conversions (cs) +import SetupHelpers +import Test.Bot (mkBotService) +import Test.Migration.Util (waitForMigration) +import Testlib.MockIntegrationService (MockServerSettings, withMockServer) +import Testlib.Prelude +import Testlib.ResourcePool + +-- | Migrate the 'ServiceStore' (old bot provider/service connection data) from +-- Cassandra to PostgreSQL. A service is created while galley reads from +-- Cassandra; after the background worker backfills it, the service must still be +-- usable once galley reads exclusively from PostgreSQL (verified by adding a bot +-- to a conversation, which forces galley to look up the service). +testServiceMigration :: (HasCallStack) => App () +testServiceMigration = do + resourcePool <- asks (.resourcePool) + let settings = def :: MockServerSettings + runCodensity (acquireResources 1 resourcePool) $ \[backend] -> do + let domain = backend.berDomain + -- The mock service must stay alive across all lifecycle stages, since adding + -- a bot in the final (PostgreSQL) stage requires the service endpoint to be + -- reachable. + withMockServer settings mkBotService $ \(host, port) _chan -> do + let serviceUrl = "https://" <> host <> ":" <> show port + (alice, providerId, serviceId) <- + runCodensity (startDynamicBackend backend (conf "cassandra" False)) $ \_ -> do + alice <- randomUser OwnDomain def + password <- randomString 20 + provider <- setupProvider alice def {newProviderPassword = Just password} + pid <- provider %. "id" & asString + service <- + newService OwnDomain pid + $ def {newServiceUrl = serviceUrl, newServiceKey = cs settings.publicKey} + sid <- asString $ service %. "id" + assertStatus 200 =<< updateServiceConn OwnDomain pid sid (object ["enabled" .= True, "password" .= password]) + pure (alice, pid, sid) + runCodensity (startDynamicBackend backend (conf "migration-to-postgresql" True)) $ \_ -> + waitForMigration domain counterName + runCodensity (startDynamicBackend backend (conf "postgresql" False)) $ \_ -> do + conv <- getJSON 201 =<< postConversation alice defProteus + cid <- conv %. "qualified_id" & objId + bindResponse (addBot alice providerId serviceId cid) $ \res -> + res.status `shouldMatchInt` 201 + where + conf :: String -> Bool -> ServiceOverrides + conf db runMigration = + def + { galleyCfg = setField "postgresMigration.service" db, + backgroundWorkerCfg = setField "migrateService" runMigration + } + +counterName :: String +counterName = "^wire_service_migration_finished" diff --git a/libs/wire-api/src/Wire/API/PostgresMarshall.hs b/libs/wire-api/src/Wire/API/PostgresMarshall.hs index e1a6f55f18d..fb5e66dfc49 100644 --- a/libs/wire-api/src/Wire/API/PostgresMarshall.hs +++ b/libs/wire-api/src/Wire/API/PostgresMarshall.hs @@ -568,6 +568,9 @@ instance PostgresMarshall ByteString HttpsUrl where instance PostgresMarshall ByteString Token where postgresMarshall = (.unToken) +instance PostgresMarshall ByteString (Fingerprint a) where + postgresMarshall = fingerprintBytes + instance PostgresMarshall Text DnsVerificationToken where postgresMarshall = Ascii.toText . (.unDnsVerificationToken) @@ -1032,6 +1035,9 @@ instance PostgresUnmarshall ByteString HttpsUrl where instance PostgresUnmarshall ByteString Token where postgresUnmarshall = Right . Token +instance PostgresUnmarshall ByteString (Fingerprint a) where + postgresUnmarshall = Right . Fingerprint + instance PostgresUnmarshall Text DnsVerificationToken where postgresUnmarshall = first Text.pack . fmap DnsVerificationToken . Ascii.validate diff --git a/libs/wire-api/src/Wire/API/Provider/Service.hs b/libs/wire-api/src/Wire/API/Provider/Service.hs index 2329ee4c102..f67b2b7bcb1 100644 --- a/libs/wire-api/src/Wire/API/Provider/Service.hs +++ b/libs/wire-api/src/Wire/API/Provider/Service.hs @@ -53,6 +53,7 @@ where import Cassandra.CQL hiding (Set) import Control.Lens (makeLenses, (?~)) +import Data.Bifunctor (first) import Data.Aeson (FromJSON (..), ToJSON (..)) import Data.Aeson qualified as A import Data.ByteString.Builder qualified as BB @@ -74,6 +75,7 @@ import Imports import Wire.API.Provider.Service.Tag (ServiceTag (..)) import Wire.API.Routes.MultiVerb import Wire.API.User.Profile (Asset, Name) +import Wire.API.PostgresMarshall import Wire.Arbitrary (Arbitrary (arbitrary), GenericUniform (..)) -------------------------------------------------------------------------------- @@ -284,6 +286,12 @@ instance S.ToSchema ServiceToken where deriving instance Cql ServiceToken +instance PostgresMarshall ByteString ServiceToken where + postgresMarshall = toByteString' + +instance PostgresUnmarshall ByteString ServiceToken where + postgresUnmarshall = first Text.pack . runParser parser + -------------------------------------------------------------------------------- -- ServiceProfile diff --git a/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/PostgresMarshall.hs b/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/PostgresMarshall.hs index 191f55bffd0..b2e5c9bbbdc 100644 --- a/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/PostgresMarshall.hs +++ b/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/PostgresMarshall.hs @@ -23,7 +23,7 @@ import Crypto.KDF.Argon2 qualified as Argon2 import Data.Aeson as A import Data.ByteString.Char8 qualified as BS8 import Data.Code qualified as Code -import Data.Misc (PlainTextPassword8, fromPlainTextPassword) +import Data.Misc (Fingerprint, PlainTextPassword8, Rsa, fromPlainTextPassword) import Data.Text.Encoding (encodeUtf8) import Imports import Test.Tasty qualified as T @@ -33,6 +33,7 @@ import Wire.API.Password as Password import Wire.API.Password.Argon2id (Argon2HashedPassword (..), encodeArgon2HashedPassword) import Wire.API.Password.Scrypt (encodeScryptPassword) import Wire.API.PostgresMarshall +import Wire.API.Provider.Service (ServiceToken) import Wire.API.Team.Feature import Wire.Arbitrary qualified as Arbitrary () @@ -44,7 +45,9 @@ tests = testRoundTrip @ByteString @Password.Password, testRoundTrip @Int32 @FeatureStatus, testRoundTrip @Int32 @LockStatus, - testRoundTrip @A.Value @DbConfig + testRoundTrip @A.Value @DbConfig, + testRoundTrip @ByteString @(Fingerprint Rsa), + testRoundTrip @ByteString @ServiceToken ] testRoundTrip :: diff --git a/libs/wire-subsystems/postgres-migrations/20260804125818-service.sql b/libs/wire-subsystems/postgres-migrations/20260804125818-service.sql new file mode 100644 index 00000000000..deb5c308f58 --- /dev/null +++ b/libs/wire-subsystems/postgres-migrations/20260804125818-service.sql @@ -0,0 +1,9 @@ +CREATE TABLE IF NOT EXISTS service ( + provider uuid NOT NULL, + id uuid NOT NULL, + base_url bytea NOT NULL, + auth_token bytea NOT NULL, + fingerprints bytea[] NOT NULL DEFAULT '{}', + enabled boolean NOT NULL, + PRIMARY KEY (provider, id) +); diff --git a/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs b/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs index 327862f7cd5..c07e42e6273 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, + service :: StorageLocation } deriving (Show) @@ -68,3 +69,4 @@ instance FromJSON PostgresMigrationOpts where <*> o .: "teamFeatures" <*> o .: "domainRegistration" <*> o .: "user" + <*> o .: "service" diff --git a/libs/wire-subsystems/src/Wire/ServiceStore/Cassandra/Queries.hs b/libs/wire-subsystems/src/Wire/ServiceStore/Cassandra/Queries.hs new file mode 100644 index 00000000000..a465ed69ed8 --- /dev/null +++ b/libs/wire-subsystems/src/Wire/ServiceStore/Cassandra/Queries.hs @@ -0,0 +1,27 @@ +-- 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.ServiceStore.Cassandra.Queries where + +import Cassandra as C hiding (Value) +import Data.Id +import Data.Misc (Fingerprint, HttpsUrl, Rsa) +import Imports +import Wire.API.Provider.Service (ServiceToken) + +selectAllServices :: PrepQuery R () (ProviderId, ServiceId, HttpsUrl, ServiceToken, C.Set (Fingerprint Rsa), Bool) +selectAllServices = "SELECT provider, id, base_url, auth_token, fingerprints, enabled FROM service" diff --git a/libs/wire-subsystems/src/Wire/ServiceStore/DualWrite.hs b/libs/wire-subsystems/src/Wire/ServiceStore/DualWrite.hs new file mode 100644 index 00000000000..052a6338662 --- /dev/null +++ b/libs/wire-subsystems/src/Wire/ServiceStore/DualWrite.hs @@ -0,0 +1,49 @@ +-- 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.ServiceStore.DualWrite + ( interpretServiceStoreToCassandraAndPostgres, + ) +where + +import Cassandra (ClientState) +import Imports +import Polysemy +import Polysemy.TinyLog (TinyLog) +import Wire.Postgres (PGConstraints) +import Wire.ServiceStore (ServiceStore (..)) +import Wire.ServiceStore qualified as ServiceStore +import Wire.ServiceStore.Cassandra qualified as Cassandra +import Wire.ServiceStore.Postgres qualified as Postgres + +-- | Cassandra is the source of truth during migration; writes are mirrored to Postgres. +interpretServiceStoreToCassandraAndPostgres :: + ( Member TinyLog r, + PGConstraints r + ) => + ClientState -> + Sem (ServiceStore ': r) a -> + Sem r a +interpretServiceStoreToCassandraAndPostgres cassClient = interpret $ \case + GetService sr -> + Cassandra.interpretServiceStoreToCassandra cassClient $ ServiceStore.getService sr + CreateService s -> do + Cassandra.interpretServiceStoreToCassandra cassClient $ ServiceStore.createService s + Postgres.interpretServiceStoreToPostgres $ ServiceStore.createService s + DeleteService sr -> do + Cassandra.interpretServiceStoreToCassandra cassClient $ ServiceStore.deleteService sr + Postgres.interpretServiceStoreToPostgres $ ServiceStore.deleteService sr diff --git a/libs/wire-subsystems/src/Wire/ServiceStore/Migration.hs b/libs/wire-subsystems/src/Wire/ServiceStore/Migration.hs new file mode 100644 index 00000000000..4628794e367 --- /dev/null +++ b/libs/wire-subsystems/src/Wire/ServiceStore/Migration.hs @@ -0,0 +1,159 @@ +-- 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.ServiceStore.Migration (migrateServicesLoop) where + +import Cassandra hiding (Value) +import Cassandra qualified as C +import Data.ByteString.Conversion (toByteString') +import Data.Conduit +import Data.Conduit.List qualified as C +import Data.IORef qualified as IORef +import Data.Id +import Data.Misc (Fingerprint, HttpsUrl, Rsa) +import Data.Text qualified as T +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.API.Bot.Service qualified as Bot +import Wire.API.Provider.Service (ServiceToken, newServiceRef) +import Wire.Migration +import Wire.Postgres +import Wire.Sem.Logger (mapLogger) +import Wire.Sem.Logger.TinyLog (loggerToTinyLog) +import Wire.ServiceStore (createService) +import Wire.ServiceStore.Cassandra.Queries qualified as Cql +import Wire.ServiceStore.Postgres qualified as Postgres + +type EffectStack = + [ State Int, + Input ClientState, + Input Hasql.Pool, + Resource, + Async, + Race, + TinyLog, + Embed IO, + Final IO + ] + +migrateServicesLoop :: + MigrationOptions -> + ClientState -> + Hasql.Pool -> + Log.Logger -> + Prometheus.Counter -> + Prometheus.Counter -> + Prometheus.Counter -> + Prometheus.Vector Text Prometheus.Histogram -> + IO () +migrateServicesLoop migOpts cassClient pgPool logger migCounter migFinished migFailed migDuration = + migrationLoop + logger + "services" + migFinished + migFailed + (interpreter cassClient pgPool logger "services") + (migrateAllServices 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 + +migrateAllServices :: + ( 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) () +migrateAllServices migOpts migCounter migDuration = do + lift $ info $ Log.msg (Log.val "migrateAllServices") + withCount (paginateSem Cql.selectAllServices (paramsP LocalQuorum () migOpts.pageSize) x5) + .| logRetrievedPage migOpts.pageSize id + .| C.mapM_ (traverse_ (\row@(pid, sid, _, _, _, _) -> handleErrors (toByteString' pid <> toByteString' sid) (migrateServiceRow migOpts migCounter migDuration row))) + +migrateServiceRow :: + ( PGConstraints r, + Member TinyLog r, + Member Resource r, + Member Race r + ) => + MigrationOptions -> + Prometheus.Counter -> + Prometheus.Vector Text Prometheus.Histogram -> + (ProviderId, ServiceId, HttpsUrl, ServiceToken, C.Set (Fingerprint Rsa), Bool) -> + Sem r () +migrateServiceRow migOpts migCounter migDuration (pid, sid, url, tok, Set fps, ena) = do + let service = Bot.Service (newServiceRef sid pid) url tok fps ena + keyText = T.pack (show (pid, sid)) + outcomeRef <- liftIO $ IORef.newIORef @Text "error" + bracket + (liftIO getCurrentTime) + (observeDuration migDuration outcomeRef) + ( const $ do + timeoutResult <- Conc.timeout (migOpts.timeout <$ handleTimeout) migOpts.timeout $ Postgres.interpretServiceStoreToPostgres $ createService service + case timeoutResult of + Left timedOutAfter -> do + markOutcome outcomeRef "timeout" + liftIO . UnliftIO.throwIO $ MigrationTimedOut keyText timedOutAfter + Right () -> do + markOutcome outcomeRef "success" + liftIO $ Prometheus.incCounter migCounter + ) + where + handleTimeout = + err $ + Log.msg (Log.val "service migration timed out") + . Log.field "provider" (show pid) + . Log.field "service" (show sid) + . 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/ServiceStore/Postgres.hs b/libs/wire-subsystems/src/Wire/ServiceStore/Postgres.hs new file mode 100644 index 00000000000..5fd1f3cffd8 --- /dev/null +++ b/libs/wire-subsystems/src/Wire/ServiceStore/Postgres.hs @@ -0,0 +1,105 @@ +-- 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.ServiceStore.Postgres + ( interpretServiceStoreToPostgres, + ) +where + +import Control.Lens +import Data.Id +import Data.Misc (Fingerprint, HttpsUrl, Rsa) +import Data.UUID (UUID) +import Data.Vector (Vector) +import Hasql.Statement qualified as Hasql +import Hasql.TH +import Imports +import Polysemy +import Wire.API.Bot.Service qualified as Bot +import Wire.API.PostgresMarshall +import Wire.API.Provider.Service (ServiceRef, ServiceToken, serviceRefId, serviceRefProvider) +import Wire.Postgres +import Wire.ServiceStore (ServiceStore (..)) + +interpretServiceStoreToPostgres :: + (PGConstraints r) => + Sem (ServiceStore ': r) a -> + Sem r a +interpretServiceStoreToPostgres = interpret $ \case + CreateService s -> insertService s + GetService sr -> lookupService sr + DeleteService sr -> deleteService sr + +insertService :: (PGConstraints r) => Bot.Service -> Sem r () +insertService s = + runStatement + ( s ^. Bot.serviceRef . serviceRefProvider, + s ^. Bot.serviceRef . serviceRefId, + s ^. Bot.serviceUrl, + s ^. Bot.serviceToken, + s ^. Bot.serviceFingerprints, + s ^. Bot.serviceEnabled + ) + insert + where + insert :: + Hasql.Statement (ProviderId, ServiceId, HttpsUrl, ServiceToken, [Fingerprint Rsa], Bool) () + insert = + lmapPG @(UUID, UUID, ByteString, ByteString, Vector ByteString, Bool) + [resultlessStatement|INSERT INTO service + (provider, id, base_url, auth_token, fingerprints, enabled) + VALUES + ($1 :: uuid, $2 :: uuid, $3 :: bytea, $4 :: bytea, $5 :: bytea[], $6 :: boolean) + ON CONFLICT (provider, id) DO UPDATE + SET base_url = ($3 :: bytea), + auth_token = ($4 :: bytea), + fingerprints = ($5 :: bytea[]), + enabled = ($6 :: boolean) + |] + +lookupService :: + (PGConstraints r) => + ServiceRef -> + Sem r (Maybe Bot.Service) +lookupService sr = + fmap (\(url, tok, fps, ena) -> Bot.Service sr url tok fps ena) + <$> runStatement (sr ^. serviceRefProvider, sr ^. serviceRefId) select + where + select :: + Hasql.Statement + (ProviderId, ServiceId) + (Maybe (HttpsUrl, ServiceToken, [Fingerprint Rsa], Bool)) + select = + dimapPG @(UUID, UUID) @(ProviderId, ServiceId) @(Maybe (ByteString, ByteString, Vector ByteString, Bool)) + [maybeStatement|SELECT (base_url :: bytea), + (auth_token :: bytea), + (fingerprints :: bytea[]), + (enabled :: boolean) + FROM service + WHERE provider = ($1 :: uuid) AND id = ($2 :: uuid) + |] + +deleteService :: (PGConstraints r) => ServiceRef -> Sem r () +deleteService sr = + runStatement (sr ^. serviceRefProvider, sr ^. serviceRefId) delete + where + delete :: Hasql.Statement (ProviderId, ServiceId) () + delete = + lmapPG + [resultlessStatement|DELETE FROM service + WHERE provider = ($1 :: uuid) AND id = ($2 :: uuid) + |] diff --git a/libs/wire-subsystems/wire-subsystems.cabal b/libs/wire-subsystems/wire-subsystems.cabal index 943f8fa0294..065b66cc049 100644 --- a/libs/wire-subsystems/wire-subsystems.cabal +++ b/libs/wire-subsystems/wire-subsystems.cabal @@ -443,6 +443,10 @@ library Wire.ScimUserTimesStore.Mem Wire.ServiceStore Wire.ServiceStore.Cassandra + Wire.ServiceStore.Cassandra.Queries + Wire.ServiceStore.DualWrite + Wire.ServiceStore.Migration + Wire.ServiceStore.Postgres Wire.SessionStore Wire.SessionStore.Cassandra Wire.SFT diff --git a/postgres-schema.sql b/postgres-schema.sql index 948f1d62594..25969ff82e9 100644 --- a/postgres-schema.sql +++ b/postgres-schema.sql @@ -9,8 +9,8 @@ \restrict 79bbfb4630959c48307653a5cd3d83f2582b3c2210f75f10d79e3ebf0015620 --- Dumped from database version 17.9 --- Dumped by pg_dump version 17.9 +-- Dumped from database version 17.10 +-- Dumped by pg_dump version 17.10 SET statement_timeout = 0; SET lock_timeout = 0; @@ -374,6 +374,22 @@ CREATE TABLE public.schema_migrations ( ALTER TABLE public.schema_migrations OWNER TO "wire-server"; +-- +-- Name: service; Type: TABLE; Schema: public; Owner: wire-server +-- + +CREATE TABLE public.service ( + provider uuid NOT NULL, + id uuid NOT NULL, + base_url bytea NOT NULL, + auth_token bytea NOT NULL, + fingerprints bytea[] DEFAULT '{}'::bytea[] NOT NULL, + enabled boolean NOT NULL +); + + +ALTER TABLE public.service OWNER TO "wire-server"; + -- -- Name: subconversation; Type: TABLE; Schema: public; Owner: wire-server -- @@ -609,6 +625,14 @@ ALTER TABLE ONLY public.remote_conversation_local_member ADD CONSTRAINT remote_conversation_local_member_pkey PRIMARY KEY ("user", conv_remote_domain, conv_remote_id); +-- +-- Name: service service_pkey; Type: CONSTRAINT; Schema: public; Owner: wire-server +-- + +ALTER TABLE ONLY public.service + ADD CONSTRAINT service_pkey PRIMARY KEY (provider, id); + + -- -- Name: subconversation subconversation_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..5f42d7760d1 100644 --- a/services/background-worker/background-worker.integration.yaml +++ b/services/background-worker/background-worker.integration.yaml @@ -92,3 +92,4 @@ postgresMigration: teamFeatures: postgresql domainRegistration: postgresql user: postgresql + service: postgresql diff --git a/services/background-worker/src/Wire/BackgroundWorker.hs b/services/background-worker/src/Wire/BackgroundWorker.hs index b57ba12df40..5ab8d6e6563 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 () + + cleanupServiceMigration <- + if opts.migrateService + then + runAppT env $ + withNamedLogger "migrate-service" $ + Migrations.service 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 cleanupServiceMigration <*> 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..387050dbb83 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, + migrateService :: !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..2e8b9e6ccdd 100644 --- a/services/background-worker/src/Wire/PostgresMigrations.hs +++ b/services/background-worker/src/Wire/PostgresMigrations.hs @@ -27,6 +27,7 @@ import Wire.CodeStore.Migration import Wire.ConversationStore.Migration import Wire.DomainRegistrationStore.Migration import Wire.Migration (MigrationOptions) +import Wire.ServiceStore.Migration import Wire.TeamFeatureStore.Migration conversations :: MigrationOptions -> AppT IO CleanupAction @@ -107,3 +108,21 @@ domainRegistration migOpts = do pure $ do Log.info logger $ Log.msg (Log.val "cancelling domain registration migration") cancel migrationLoop + +service :: MigrationOptions -> AppT IO CleanupAction +service migOpts = do + cassClient <- asks (.cassandraGalley) + pgPool <- asks (.hasqlPool) + logger <- asks (.logger) + Log.info logger $ Log.msg (Log.val "starting service migration") + count <- register $ counter $ Prometheus.Info "wire_service_migrated_to_pg" "Number of services migrated to Postgresql" + finished <- register $ counter $ Prometheus.Info "wire_service_migration_finished" "Whether the service migration to Postgresql is finished successfully" + failed <- register $ counter $ Prometheus.Info "wire_service_migration_failed" "Whether the service migration to Postgresql has failed" + duration <- register $ vector "outcome" $ histogram (Prometheus.Info "wire_service_migration_duration_seconds" "Duration of service migration attempts") defaultBuckets + + migrationLoop <- async . lift $ migrateServicesLoop migOpts cassClient pgPool logger count finished failed duration + + Log.info logger $ Log.msg (Log.val "started service migration") + pure $ do + Log.info logger $ Log.msg (Log.val "cancelling service 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..92130e332ac 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, + service = CassandraStorage } gundeckEndpoint = undefined brigEndpoint = undefined @@ -435,7 +436,8 @@ spec = do conversationCodes = CassandraStorage, teamFeatures = CassandraStorage, domainRegistration = CassandraStorage, - user = CassandraStorage + user = CassandraStorage, + service = 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..de165afbcb1 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, + service = CassandraStorage } statuses <- newIORef mempty backendNotificationMetrics <- mkBackendNotificationMetrics diff --git a/services/brig/brig.integration.yaml b/services/brig/brig.integration.yaml index 8be11f028bd..f1980325995 100644 --- a/services/brig/brig.integration.yaml +++ b/services/brig/brig.integration.yaml @@ -176,6 +176,7 @@ postgresMigration: teamFeatures: postgresql domainRegistration: postgresql user: postgresql + service: postgresql optSettings: setActivationTimeout: 4 diff --git a/services/galley/galley.integration.yaml b/services/galley/galley.integration.yaml index 47980040c1e..cc04d025509 100644 --- a/services/galley/galley.integration.yaml +++ b/services/galley/galley.integration.yaml @@ -266,3 +266,4 @@ postgresMigration: teamFeatures: postgresql domainRegistration: postgresql user: postgresql + service: postgresql diff --git a/services/galley/src/Galley/App.hs b/services/galley/src/Galley/App.hs index bd0131b4d2d..32fa1145177 100644 --- a/services/galley/src/Galley/App.hs +++ b/services/galley/src/Galley/App.hs @@ -164,6 +164,8 @@ import Wire.Sem.Random (Random) import Wire.Sem.Random.IO import Wire.ServiceStore (ServiceStore) import Wire.ServiceStore.Cassandra (interpretServiceStoreToCassandra) +import Wire.ServiceStore.DualWrite (interpretServiceStoreToCassandraAndPostgres) +import Wire.ServiceStore.Postgres (interpretServiceStoreToPostgres) import Wire.SparAPIAccess (SparAPIAccess) import Wire.SparAPIAccess.Rpc import Wire.TeamCollaboratorsStore (TeamCollaboratorsStore) @@ -423,6 +425,11 @@ evalGalley e = CassandraStorage -> interpretTeamFeatureStoreToCassandra MigrationToPostgresql -> interpretTeamFeatureStoreToCassandraAndPostgres PostgresqlStorage -> interpretTeamFeatureStoreToPostgres + serviceStoreInterpreter = + case (e ^. options . postgresMigration).service of + CassandraStorage -> interpretServiceStoreToCassandra (e ^. cstate) + MigrationToPostgresql -> interpretServiceStoreToCassandraAndPostgres (e ^. cstate) + PostgresqlStorage -> interpretServiceStoreToPostgres localUnit = toLocalUnsafe (e ^. options . settings . federationDomain) () teamSubsystemConfig = TeamSubsystemConfig @@ -528,7 +535,7 @@ evalGalley e = . interpretMLSCommitLockStoreToCassandra (e ^. cstate) . convStoreInterpreter . interpretTeamNotificationStoreToCassandra - . interpretServiceStoreToCassandra (e ^. cstate) + . serviceStoreInterpreter . interpretUserGroupStoreToPostgres . runInputConst legalHoldEnv . interpretLegalHoldStoreToCassandra lh