WPB-22953: migrate connections to PostGreSQL

changelog.d/5-internal/WPB-22953

@@ -0,0 +1 @@
+Migration of connections from cassandra to postgres

charts/wire-server/templates/background-worker/configmap.yaml

@@ -89,6 +89,7 @@ data:
     migratePasswordReset: {{ .migratePasswordReset }}
     migrateCustomBackend: {{ .migrateCustomBackend }}
     migrateNonce: {{ .migrateNonce }}
+    migrateConnections: {{ .migrateConnections }}
     migrationOptions:
 {{ toYaml .migrationOptions | indent 6 }}
 

charts/wire-server/values.yaml

@@ -98,6 +98,7 @@ galley:
       customBackend: cassandra
       budget: cassandra
       nonce: cassandra
+      connections: cassandra
     settings:
       httpPoolSize: 128
       maxTeamSize: 10000
@@ -1049,6 +1050,10 @@ background-worker:
     # It's important to set `settings.postgresMigration.nonce` to `migration-to-postgresql`
     # before starting the migration.
     migrateNonce: false
+    # This will start the migration of connection data.
+    # It's important to set `settings.postgresMigration.connections` to `migration-to-postgresql`
+    # before starting the migration.
+    migrateConnections: false
 
     # This will start the migration of custom backends. It's important to set
     # galley's postgresMigration.customBackend to migration-to-postgresql first.

docs/src/developer/reference/config-options.md

@@ -2334,6 +2334,7 @@ galley:
       customBackend: postgresql
       budget: postgresql
       nonce: postgresql
+      connections: postgresql
 background-worker:
   config:
     migrateConversations: false
@@ -2344,6 +2345,7 @@ background-worker:
     migratePasswordReset: false
     migrateCustomBackend: false
     migrateNonce: false
+    migrateConnections: false
 ```
 
 #### Migration for existing installations
@@ -2379,6 +2381,7 @@ The current settings and their background-worker flags are:
 - `passwordReset` -> `migratePasswordReset`
 - `customBackend` -> `migrateCustomBackend`
 - `budget` -> none (ephemeral TTL'd rate-limit counters; no backfill worker, no `wire_budget_migration_finished` metric — Cassandra rows age out on their own while new writes go to PostgreSQL under `migration-to-postgresql`). Because there is no backfill, keep `budget` on `migration-to-postgresql` for at least the longest configured `limitFailedLogins.timeout` (with PostgreSQL mirror writes healthy) before cutting over to `postgresql`; otherwise counters that only exist in Cassandra are dropped and affected users get one extra rate-limit window.
+- `connections` -> `migrateConnections`
 
 **Migration pattern per migration setting**
 
@@ -2404,6 +2407,7 @@ The current settings and their background-worker flags are:
          customBackend: cassandra
          budget: cassandra
          nonce: cassandra
+         connections: cassandra
    background-worker:
      config:
        migrateConversations: false
@@ -2415,6 +2419,7 @@ The current settings and their background-worker flags are:
        migratePasswordReset: false
        migrateCustomBackend: false
        migrateNonce: false
+       migrateConnections: false
    ```
 
    This change should restart the affected pods, and new writes will follow the
@@ -2433,6 +2438,7 @@ The current settings and their background-worker flags are:
        migrateBlockList: true
        migratePasswordReset: true
        migrateCustomBackend: true
+       migrateConnections: true
    ```
 
    During migration, Cassandra rows are not deleted. Writes and migration share
@@ -2462,6 +2468,7 @@ The current settings and their background-worker flags are:
    - `passwordReset`: `wire_password_reset_migration_finished`
    - `customBackend`: `wire_custom_backend_migration_finished`
    - `nonce`: `wire_nonce_migration_finished`
+   - `connections`: `wire_connections_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
@@ -2481,6 +2488,7 @@ The current settings and their background-worker flags are:
          customBackend: postgresql
          budget: postgresql
          nonce: postgresql
+         connections: postgresql
    background-worker:
      config:
        migrateConversations: false
@@ -2492,6 +2500,7 @@ The current settings and their background-worker flags are:
        migratePasswordReset: false
        migrateCustomBackend: false
        migrateNonce: false
+       migrateConnections: false
    ```
 
 **How to run migrations independently or in batches**
@@ -2584,6 +2593,7 @@ migrateBlockList: false
 migratePasswordReset: false
 migrateCustomBackend: false
 migrateNonce: false
+migrateConnections: false
 
 # migration settings
 migrationOptions:

hack/helm_vars/common.yaml.gotmpl

@@ -24,6 +24,7 @@ userStore: {{ $preferredStore }}
 blockListStore: {{ $preferredStore }}
 budgetStore: {{ $preferredStore }}
 nonceStore: {{ $preferredStore }}
+connectionsStore: {{ $preferredStore }}
 
 {{- if (eq (env "UPLOAD_XML_S3_BASE_URL") "") }}
 uploadXml: {}

hack/helm_vars/wire-server/values.yaml.gotmpl

@@ -315,6 +315,7 @@ galley:
       blockList: {{ .Values.blockListStore }}
       budget: {{ .Values.budgetStore }}
       nonce: {{ .Values.nonceStore }}
+      connections: {{ .Values.connectionsStore }}
     settings:
       maxConvAndTeamSize: 16
       maxTeamSize: 32

integration/integration.cabal

@@ -179,6 +179,7 @@ library
     Test.MessageTimer
     Test.Migration.BlockList
     Test.Migration.Budget
+    Test.Migration.Connections
     Test.Migration.Conversation
     Test.Migration.ConversationCodes
     Test.Migration.CustomBackend

integration/test/Test/Migration/Connections.hs

@@ -0,0 +1,107 @@
+-- This file is part of the Wire Server implementation.
+--
+-- Copyright (C) 2026 Wire Swiss GmbH <opensource@wire.com>
+--
+-- This program is free software: you can redistribute it and/or modify it under
+-- the terms of the GNU Affero General Public License as published by the Free
+-- Software Foundation, either version 3 of the License, or (at your option) any
+-- later version.
+--
+-- This program is distributed in the hope that it will be useful, but WITHOUT
+-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
+-- details.
+--
+-- You should have received a copy of the GNU Affero General Public License along
+-- with this program. If not, see <https://www.gnu.org/licenses/>.
+
+module Test.Migration.Connections (testConnectionsMigrationToPostgres) where
+
+import API.Brig (getConnection, postConnection, putConnection)
+import Control.Monad.Codensity
+import Control.Monad.Reader
+import GHC.Stack
+import SetupHelpers
+import Test.Migration.Util (assertMigrationSuccessful, waitForMigration)
+import Testlib.Prelude
+import Testlib.ResourcePool
+
+-- | Drives the connections store through the full cutover lifecycle
+-- (cassandra -> migration-to-postgresql -> postgresql) on a dynamic backend.
+-- Every phase creates a fresh pair of users and connects them via the brig
+-- public API, exercising the writes of all three interpreters (Cassandra,
+-- dual-write, Postgres). Connections written by earlier interpreters must
+-- remain readable afterwards: the pair connected during the pure-Cassandra
+-- phase proves the background worker backfilled rows that only ever existed
+-- in Cassandra.
+--
+-- Remote ('connection_remote') rows are not seeded explicitly here; the
+-- migration's copy loop still runs over whatever dual-write produced, and
+-- federated connections are covered by the brig integration suite.
+testConnectionsMigrationToPostgres :: (HasCallStack) => App ()
+testConnectionsMigrationToPostgres = do
+  resourcePool <- asks (.resourcePool)
+  runCodensity (acquireResources 1 resourcePool) $ \[backend] -> do
+    let domain = backend.berDomain
+
+    -- P1 cassandra: connect a pair via the pure Cassandra interpreter
+    cassandraPair <-
+      runCodensity (startDynamicBackend backend (conf "cassandra" False)) $ \_ ->
+        connectFreshPair domain
+
+    -- P2 migration-to-postgresql (worker off): reads still come from
+    -- Cassandra, writes go to both stores
+    dualWritePair <-
+      runCodensity (startDynamicBackend backend (conf "migration-to-postgresql" False)) $ \_ -> do
+        checkConnection cassandraPair
+        connectFreshPair domain
+
+    -- P3 migration-to-postgresql (worker on): dual-write while backfilling
+    workerOnPair <-
+      runCodensity (startDynamicBackend backend (conf "migration-to-postgresql" True)) $ \_ -> do
+        checkConnection dualWritePair
+        pair <- connectFreshPair domain
+        waitForMigration domain connectionsMigrationFinishedCounterName
+        -- The failed counter is only registered while the migration worker
+        -- runs, so assert it in this phase (worker on), not in P4.
+        assertMigrationSuccessful domain "^wire_connections_migration_failed"
+        pure pair
+
+    -- P4 postgresql: reads are served exclusively from Postgres
+    runCodensity (startDynamicBackend backend (conf "postgresql" False)) $ \_ -> do
+      -- Connections written by every interpreter are visible to the pure
+      -- Postgres interpreter.
+      checkConnection cassandraPair
+      checkConnection dualWritePair
+      checkConnection workerOnPair
+
+      -- Write and read via the pure Postgres interpreter.
+      pgPair <- connectFreshPair domain
+      checkConnection pgPair
+  where
+    conf db runMigration =
+      def
+        { brigCfg = setField "postgresMigration.connections" db,
+          backgroundWorkerCfg =
+            setField "postgresMigration.connections" db
+              >=> setField "migrateConnections" runMigration
+        }
+
+connectionsMigrationFinishedCounterName :: String
+connectionsMigrationFinishedCounterName = "^wire_connections_migration_finished"
+
+-- | Create two fresh users on the given domain and connect them (request +
+-- accept) via the brig public connection API.
+connectFreshPair :: (HasCallStack, MakesValue domain) => domain -> App (Value, Value)
+connectFreshPair domain = do
+  [u1, u2] <- createUsers [domain, domain]
+  void $ postConnection u1 u2 >>= assertSuccess
+  void $ putConnection u2 u1 "accepted" >>= assertSuccess
+  pure (u1, u2)
+
+-- | The connection between the two users is visible with status 'accepted'.
+checkConnection :: (HasCallStack) => (Value, Value) -> App ()
+checkConnection (u1, u2) =
+  getConnection u1 u2 `bindResponse` \resp -> do
+    resp.status `shouldMatchInt` 200
+    resp.json %. "status" `shouldMatch` "accepted"

integration/test/Testlib/ModService.hs

@@ -63,6 +63,7 @@ import qualified Data.Yaml as Yaml
 import GHC.Stack
 import qualified Network.HTTP.Client as HTTP
 import System.Directory (copyFile, createDirectoryIfMissing, doesDirectoryExist, doesFileExist, listDirectory, removeDirectoryRecursive, removeFile)
+import System.Environment (lookupEnv)
 import System.Exit
 import System.FilePath
 import System.IO
@@ -694,6 +695,18 @@ warmupFederation = do
             [(env.domain1, env.domain2), (env.domain2, env.domain1)]
   retryRequestUntil checkBoth "Static federator ingress (domain1 <-> domain2)"
 
+  -- The legacy fed-v* backends (ENABLE_FEDERATION_V0..V3) are static
+  -- deployments not covered by the warmup above; the first test burst against
+  -- them can fail with transient federator-transport errors (521 connection
+  -- refused, 525 SSL, 533 unreachable backend). Warm them from the local
+  -- domain before the pool starts, same doctrine as the static warmup above.
+  for_ (zip [0 :: Int ..] [env.federationV0Domain, env.federationV1Domain, env.federationV2Domain, env.federationV3Domain]) $ \(n, fedDomain) -> do
+    enabled <- liftIO (lookupEnv ("ENABLE_FEDERATION_V" <> show n))
+    when (enabled == Just "1") $
+      retryRequestUntil
+        (checkFederationIngress env.domain1 fedDomain)
+        ("Legacy federator ingress (domain1 -> fed-v" <> show n <> ")")
+
 retryRequestUntil :: (HasCallStack) => ((HasCallStack) => App Bool) -> String -> App ()
 retryRequestUntil = retryRequestUntilDebug federatorIngressDelay Nothing
 

libs/wire-api/src/Wire/API/Connection.hs

@@ -53,6 +53,7 @@ import Data.Schema
 import Data.Text as Text hiding (show)
 import Imports
 import Servant.API
+import Wire.API.PostgresMarshall (PostgresMarshall (..), PostgresUnmarshall (..))
 import Wire.API.Routes.MultiTablePaging
 import Wire.Arbitrary (Arbitrary (..), GenericUniform (..))
 import Wire.Sem.Paging
@@ -261,6 +262,37 @@ instance C.Cql RelationWithHistory where
   toCql MissingLegalholdConsentFromSent = C.CqlInt 10
   toCql MissingLegalholdConsentFromCancelled = C.CqlInt 11
 
+instance PostgresMarshall Int32 RelationWithHistory where
+  postgresMarshall = \case
+    AcceptedWithHistory -> 0
+    BlockedWithHistory -> 1
+    PendingWithHistory -> 2
+    IgnoredWithHistory -> 3
+    SentWithHistory -> 4
+    CancelledWithHistory -> 5
+    MissingLegalholdConsentFromAccepted -> 6
+    MissingLegalholdConsentFromBlocked -> 7
+    MissingLegalholdConsentFromPending -> 8
+    MissingLegalholdConsentFromIgnored -> 9
+    MissingLegalholdConsentFromSent -> 10
+    MissingLegalholdConsentFromCancelled -> 11
+
+instance PostgresUnmarshall Int32 RelationWithHistory where
+  postgresUnmarshall = \case
+    0 -> Right AcceptedWithHistory
+    1 -> Right BlockedWithHistory
+    2 -> Right PendingWithHistory
+    3 -> Right IgnoredWithHistory
+    4 -> Right SentWithHistory
+    5 -> Right CancelledWithHistory
+    6 -> Right MissingLegalholdConsentFromAccepted
+    7 -> Right MissingLegalholdConsentFromBlocked
+    8 -> Right MissingLegalholdConsentFromPending
+    9 -> Right MissingLegalholdConsentFromIgnored
+    10 -> Right MissingLegalholdConsentFromSent
+    11 -> Right MissingLegalholdConsentFromCancelled
+    n -> Left $ "unexpected RelationWithHistory: " <> Text.pack (show n)
+
 ----------------
 -- Requests
 

libs/wire-api/test/unit/Test/Wire/API/Roundtrip/PostgresMarshall.hs

@@ -46,6 +46,7 @@ import Imports
 import Test.Tasty qualified as T
 import Test.Tasty.QuickCheck
 import Type.Reflection (typeRep)
+import Wire.API.Connection qualified as Connection
 import Wire.API.Password as Password
 import Wire.API.Password.Argon2id (Argon2HashedPassword (..), encodeArgon2HashedPassword)
 import Wire.API.Password.Scrypt (encodeScryptPassword)
@@ -64,6 +65,7 @@ tests =
       testRoundTrip @ByteString @Password.Password,
       testRoundTrip @Int32 @FeatureStatus,
       testRoundTrip @Int32 @LockStatus,
+      testRoundTrip @Int32 @Connection.RelationWithHistory,
       testRoundTrip @A.Value @DbConfig
     ]
 

libs/wire-subsystems/postgres-migrations/20261021102943-connections.sql

@@ -0,0 +1,39 @@
+-- This file is part of the Wire Server implementation.
+--
+-- Copyright (C) 2026 Wire Swiss GmbH <opensource@wire.com>
+--
+-- This program is free software: you can redistribute it and/or modify it under
+-- the terms of the GNU Affero General Public License as published by the Free
+-- Software Foundation, either version 3 of the License, or (at your option) any
+-- later version.
+--
+-- This program is distributed in the hope that it will be useful, but WITHOUT
+-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
+-- details.
+--
+-- You should have received a copy of the GNU Affero General Public License along
+-- with this program. If not, see <https://www.gnu.org/licenses/>.
+
+CREATE TABLE IF NOT EXISTS connection (
+  "left" uuid NOT NULL,
+  "right" uuid NOT NULL,
+  status int NOT NULL,
+  last_update timestamptz NOT NULL,
+  conv uuid,
+  PRIMARY KEY ("left", "right")
+);
+
+CREATE TABLE IF NOT EXISTS connection_remote (
+  "left" uuid NOT NULL,
+  right_domain text NOT NULL,
+  right_user uuid NOT NULL,
+  status int NOT NULL,
+  last_update timestamptz NOT NULL,
+  conv_domain text,
+  conv_id uuid,
+  PRIMARY KEY ("left", right_domain, right_user)
+);
+
+-- Parity with the Cassandra secondary index on connection_remote.right_domain.
+CREATE INDEX IF NOT EXISTS connection_remote_right_domain_idx ON connection_remote (right_domain);