WPB-21964: move conversation creation to wire-subsystems

charts/background-worker/templates/configmap.yaml

@@ -25,6 +25,12 @@ data:
       host: brig
       port: 8080
 
+    {{- if .enableFederation }}
+    federator:
+      host: {{ .federator.host }}
+      port: {{ .federator.port }}
+    {{- end }}
+
     gundeck:
       host: gundeck
       port: 8080
@@ -103,4 +109,4 @@ data:
     {{- if .postgresMigration }}
     postgresMigration: {{- toYaml .postgresMigration | nindent 6 }}
     {{- end }}
-  {{- end }}
+  {{- end }}

charts/background-worker/values.yaml

@@ -19,6 +19,9 @@ config:
   logLevel: Info
   logFormat: StructuredJSON
   enableFederation: false # keep in sync with brig, cargohold and galley charts' config.enableFederation as well as wire-server chart's tags.federation
+  federator:
+    host: federator
+    port: 8080
   rabbitmq:
     host: rabbitmq
     port: 5672

libs/galley-types/default.nix

@@ -12,6 +12,7 @@
 , data-default
 , errors
 , gitignoreSource
+, http-types
 , imports
 , lens
 , lib
@@ -21,6 +22,7 @@
 , types-common
 , utf8-string
 , uuid
+, wai-utilities
 , wire-api
 }:
 mkDerivation {
@@ -36,6 +38,7 @@ mkDerivation {
     crypton
     data-default
     errors
+    http-types
     imports
     lens
     memory
@@ -44,6 +47,7 @@ mkDerivation {
     types-common
     utf8-string
     uuid
+    wai-utilities
     wire-api
   ];
   license = lib.licenses.agpl3Only;

libs/galley-types/galley-types.cabal

@@ -14,8 +14,10 @@ library
   -- cabal-fmt: expand src
   exposed-modules:
     Galley.Types
+    Galley.Types.Clients
     Galley.Types.Conversations.One2One
     Galley.Types.Conversations.Roles
+    Galley.Types.Error
     Galley.Types.Teams
 
   other-modules:      Paths_galley_types
@@ -76,6 +78,7 @@ library
     , crypton
     , data-default
     , errors
+    , http-types
     , imports
     , lens                   >=4.12
     , memory
@@ -84,6 +87,7 @@ library
     , types-common           >=0.16
     , utf8-string
     , uuid
+    , wai-utilities
     , wire-api
 
   default-language:   GHC2021

services/galley/src/Galley/API/Error.hs → libs/galley-types/src/Galley/Types/Error.hs

@@ -18,7 +18,7 @@
 -- | Most of the errors thrown by galley are defined as static errors in
 -- 'Wire.API.Error.Galley' and declared as part of the API. Errors defined here
 -- are dynamic, and mostly internal.
-module Galley.API.Error
+module Galley.Types.Error
   ( -- * Internal errors
     InvalidInput (..),
     InternalError (..),

libs/galley-types/src/Galley/Types/Teams.hs

@@ -23,6 +23,7 @@ module Galley.Types.Teams
   ( GetFeatureDefaults (..),
     FeatureDefaults (..),
     FeatureFlags,
+    FanoutLimit,
     featureDefaults,
     notTeamMember,
     findTeamMember,
@@ -40,13 +41,16 @@ import Data.ByteString (toStrict)
 import Data.ByteString.UTF8 qualified as UTF8
 import Data.Default
 import Data.Id (UserId)
+import Data.Range (Range)
 import Data.SOP
 import Data.Set qualified as Set
 import Imports
 import Wire.API.Team.Feature
 import Wire.API.Team.Member
 import Wire.API.Team.Permission
 
+type FanoutLimit = Range 1 HardTruncationLimit Int32
+
 -- | Used to extract the feature config type out of 'FeatureDefaults' or
 -- related types.
 type family ConfigOf a

libs/wire-api/src/Wire/API/Routes/Public/Brig.hs

@@ -1289,7 +1289,7 @@ type ClientAPI =
 -- - MemberJoin event to self and other, if joining an existing connect conversation (via galley)
 -- - ConvCreate event to self, if creating a connect conversation (via galley)
 -- - ConvConnect event to self, in some cases (via galley),
---   for details see 'Galley.API.Create.createConnectConversation'
+--   for details see 'Wire.ConversationSubsystem.Create.createConnectConversation'
 type ConnectionAPI =
   Named
     "create-connection-unqualified"

libs/wire-subsystems/src/Wire/ConversationSubsystem.hs

@@ -24,7 +24,7 @@ import Data.Qualified
 import Data.Singletons (Sing)
 import Imports
 import Polysemy
-import Wire.API.Conversation (ExtraConversationData)
+import Wire.API.Conversation (ExtraConversationData, NewConv, NewOne2OneConv)
 import Wire.API.Conversation.Action
 import Wire.API.Event.Conversation
 import Wire.NotificationSubsystem (LocalConversationUpdate)
@@ -43,5 +43,23 @@ data ConversationSubsystem m a where
     ConversationAction (tag :: ConversationActionTag) ->
     ExtraConversationData ->
     ConversationSubsystem r LocalConversationUpdate
+  CreateGroupConversation ::
+    Local UserId ->
+    Maybe ConnId ->
+    NewConv ->
+    ConversationSubsystem m StoredConversation
+  CreateOne2OneConversation ::
+    Local UserId ->
+    ConnId ->
+    NewOne2OneConv ->
+    ConversationSubsystem m (StoredConversation, Bool)
+  CreateProteusSelfConversation ::
+    Local UserId ->
+    ConversationSubsystem m (StoredConversation, Bool)
+  CreateConnectConversation ::
+    Local UserId ->
+    Maybe ConnId ->
+    Connect ->
+    ConversationSubsystem m (StoredConversation, Bool)
 
 makeSem ''ConversationSubsystem

libs/wire-subsystems/src/Wire/ConversationSubsystem/Federation.hs

@@ -0,0 +1,112 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE OverloadedRecordDot #-}
+{-# LANGUAGE PolyKinds #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
+
+-- This file is part of the Wire Server implementation.
+--
+-- Copyright (C) 2025 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 Wire.ConversationSubsystem.Federation where
+
+import Control.Error (headMay)
+import Data.Domain (Domain)
+import Data.Qualified
+import Data.Set qualified as Set
+import Imports
+import Polysemy
+import Polysemy.Error
+import Polysemy.Input
+import Wire.API.Component (Component (..))
+import Wire.API.Conversation.Protocol (ProtocolTag)
+import Wire.API.Error.Galley (NonFederatingBackends (..), UnreachableBackends (..))
+import Wire.API.Federation.API (fedClient)
+import Wire.API.Federation.API.Brig (DomainSet (..), NonConnectedBackends (..))
+import Wire.API.Federation.Client (FederatorClient)
+import Wire.API.Federation.Error
+import Wire.API.FederationStatus
+import Wire.ConversationSubsystem.Types
+import Wire.FederationAPIAccess (FederationAPIAccess)
+import Wire.FederationAPIAccess qualified as E
+
+enforceFederationProtocol ::
+  ( Member (Error FederationError) r,
+    Member (Input ConversationSubsystemConfig) r
+  ) =>
+  ProtocolTag ->
+  [Remote ()] ->
+  Sem r ()
+enforceFederationProtocol proto domains = do
+  unless (null domains) $ do
+    mAllowedProtos <- federationProtocols <$> input
+    unless (maybe True (elem proto) mAllowedProtos) $
+      throw FederationDisabledForProtocol
+
+checkFederationStatus ::
+  ( Member (Error UnreachableBackends) r,
+    Member (Error NonFederatingBackends) r,
+    Member (FederationAPIAccess FederatorClient) r
+  ) =>
+  RemoteDomains ->
+  Sem r ()
+checkFederationStatus req = do
+  status <- getFederationStatus req
+  case status of
+    FullyConnected -> pure ()
+    NotConnectedDomains dom1 dom2 -> throw (NonFederatingBackends dom1 dom2)
+
+getFederationStatus ::
+  ( Member (Error UnreachableBackends) r,
+    Member (FederationAPIAccess FederatorClient) r
+  ) =>
+  RemoteDomains ->
+  Sem r FederationStatus
+getFederationStatus req = do
+  fmap firstConflictOrFullyConnected
+    . (ensureNoUnreachableBackends =<<)
+    $ E.runFederatedConcurrentlyEither
+      (Set.toList req.rdDomains)
+      ( \qds ->
+          fedClient @'Brig @"get-not-fully-connected-backends"
+            (DomainSet . Set.map tDomain $ void qds `Set.delete` req.rdDomains)
+      )
+
+-- | "conflict" here means two remote domains that we are connected to
+-- but are not connected to each other.
+firstConflictOrFullyConnected :: [Remote NonConnectedBackends] -> FederationStatus
+firstConflictOrFullyConnected =
+  maybe
+    FullyConnected
+    (uncurry NotConnectedDomains)
+    . headMay
+    . mapMaybe toMaybeConflict
+  where
+    toMaybeConflict :: Remote NonConnectedBackends -> Maybe (Domain, Domain)
+    toMaybeConflict r =
+      headMay (Set.toList (nonConnectedBackends (tUnqualified r))) <&> (tDomain r,)
+
+ensureNoUnreachableBackends ::
+  (Member (Error UnreachableBackends) r) =>
+  [Either (Remote e, b) a] ->
+  Sem r [a]
+ensureNoUnreachableBackends results = do
+  let (errors, values) = partitionEithers results
+  unless (null errors) $
+    throw (UnreachableBackends (map (tDomain . fst) errors))
+  pure values