Add federator to galley

libs/wire-api-federation/src/Wire/API/Federation/API/Galley.hs

@@ -0,0 +1,58 @@
+{-# LANGUAGE DerivingVia #-}
+
+-- This file is part of the Wire Server implementation.
+--
+-- Copyright (C) 2020 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.API.Federation.API.Galley where
+
+import Servant.API (JSON, Post, ReqBody, (:>))
+import Servant.API.Generic ((:-))
+
+data Api routes = Api
+  { getConversation ::
+      routes
+        :- "federation"
+        :> "conversations"
+        -- usecases:
+        -- - e.g. upon registering a new client to your account, get the list of your conversations
+        :> "list-conversations"
+        :> ReqBody '[JSON] ListConversations
+        :> Post '[JSON] ListConversationsResponse,
+    conversationMemberChange ::
+      routes
+        :- "federation"
+        :> "conversations"
+        -- for the usecase:
+        -- given alice,alice2 @A and bob @B:
+        -- alice adds Bob: /add-to-conversation(bob)@A
+        --   A -> B: check bob exists on B
+        --   A: add B to conversation database entry
+        --   A -> B: by the way, B is now in one of my conversations.
+        --   (B writes this in its DB: "Bob exists in a conversation with ID 1 in A)
+        :> "member-change"
+        :> ReqBody '[JSON] ConversationMemberChange
+        :> Post '[JSON] ConversationMemberChangeResponse
+  }
+
+-- FUTUREWORK: data types, json instances, more endpoints. See https://wearezeta.atlassian.net/wiki/spaces/CORE/pages/356090113/Federation+Galley+Conversation+API for the current list we need.
+type ConversationMemberChange = ()
+
+type ConversationMemberChangeResponse = ()
+
+type ListConversations = ()
+
+type ListConversationsResponse = ()

libs/wire-api-federation/src/Wire/API/Federation/Client.hs

@@ -3,17 +3,19 @@
 
 module Wire.API.Federation.Client where
 
-import Control.Monad.Except (MonadError (..))
+import Control.Monad.Except (ExceptT, MonadError (..), withExceptT)
 import Data.ByteString.Builder (toLazyByteString)
 import qualified Data.ByteString.Lazy as LBS
 import Data.Domain (Domain, domainText)
 import qualified Data.Text as T
 import Imports
-import Mu.GRpc.Client.TyApps (GRpcMessageProtocol (MsgProtoBuf), GRpcReply (..), GrpcClient, gRpcCall)
+import Mu.GRpc.Client.TyApps (GRpcMessageProtocol (MsgProtoBuf), GRpcReply (..), GrpcClient, gRpcCall, grpcClientConfigSimple)
 import qualified Network.HTTP.Types as HTTP
 import Servant.Client (ResponseF (..))
 import qualified Servant.Client as Servant
 import Servant.Client.Core (RequestBody (..), RequestF (..), RunClient (..))
+import Util.Options (Endpoint (..))
+import Wire.API.Federation.GRPC.Client (createGrpcClient, reason)
 import qualified Wire.API.Federation.GRPC.Types as Proto
 
 data FederatorClientEnv = FederatorClientEnv
@@ -79,6 +81,12 @@ instance (Monad m, MonadError FederationClientError m) => MonadError FederationC
   throwError = FederatorClient . throwError
   catchError (FederatorClient action) f = FederatorClient $ catchError action (runFederatorClient . f)
 
+data FederationError
+  = FederationUnavailable Text
+  | FederationNotImplemented
+  | FederationNotConfigured
+  | FederationCallFailure FederationClientError
+
 data FederationClientError
   = FederationClientInvalidMethod HTTP.Method
   | FederationClientStreamingUnsupported
@@ -89,3 +97,33 @@ data FederationClientError
 
 callRemote :: MonadIO m => GrpcClient -> Proto.ValidatedFederatedRequest -> m (GRpcReply Proto.OutwardResponse)
 callRemote fedClient call = liftIO $ gRpcCall @'MsgProtoBuf @Proto.Outward @"Outward" @"call" fedClient (Proto.validatedFederatedRequestToFederatedRequest call)
+
+class HasFederatorConfig m where
+  federatorEndpoint :: m (Maybe Endpoint)
+  federationDomain :: m Domain
+
+-- FUTUREWORK: It would be nice to share the client across all calls to
+-- federator and not call this function on every invocation of federated
+-- requests, but there are some issues in http2-client which might need some
+-- fixing first. More context here:
+-- https://github.com/lucasdicioccio/http2-client/issues/37
+-- https://github.com/lucasdicioccio/http2-client/issues/49
+mkFederatorClient ::
+  (MonadIO m, HasFederatorConfig m) =>
+  ExceptT FederationError m GrpcClient
+mkFederatorClient = do
+  mbFedEndpoint <- lift federatorEndpoint
+  Endpoint host port <- maybe (throwError FederationNotConfigured) pure mbFedEndpoint
+  let cfg = grpcClientConfigSimple (T.unpack host) (fromIntegral port) False
+  createGrpcClient cfg
+    >>= either (throwError . FederationUnavailable . reason) pure
+
+executeFederated ::
+  (MonadIO m, HasFederatorConfig m) =>
+  Domain ->
+  FederatorClient component (ExceptT FederationClientError m) a ->
+  ExceptT FederationError m a
+executeFederated targetDomain action = do
+  federatorClient <- mkFederatorClient
+  originDomain <- lift federationDomain
+  withExceptT FederationCallFailure (runFederatorClientWith federatorClient targetDomain originDomain action)

libs/wire-api-federation/wire-api-federation.cabal

@@ -4,7 +4,7 @@ cabal-version: 1.12
 --
 -- see: https://github.com/sol/hpack
 --
--- hash: d46afde4e3a98dd57f7e30f974dfcc9037f1c4f5ed087f29b3f69cd4dcf732bc
+-- hash: ad51205c57d0dcf41c68effbec318807a340ee2d2abfa805b1f921fe7e3281be
 
 name:           wire-api-federation
 version:        0.1.0
@@ -25,6 +25,7 @@ library
       Wire.API.Federation.API
       Wire.API.Federation.API.Brig
       Wire.API.Federation.API.Conversation
+      Wire.API.Federation.API.Galley
       Wire.API.Federation.Client
       Wire.API.Federation.Event
       Wire.API.Federation.GRPC.Client

services/brig/src/Brig/API/Client.hs

@@ -69,6 +69,7 @@ import Network.Wai.Utilities
 import System.Logger.Class (field, msg, val, (~~))
 import qualified System.Logger.Class as Log
 import UnliftIO.Async (Concurrently (Concurrently, runConcurrently))
+import Wire.API.Federation.Client (FederationError (..))
 import qualified Wire.API.Message as Message
 import Wire.API.User.Client (QualifiedUserClientMap (..), QualifiedUserClients (..))
 import Wire.API.UserMap (QualifiedUserMap (QualifiedUserMap))

services/brig/src/Brig/API/Error.hs

@@ -37,7 +37,7 @@ import Network.HTTP.Types.Status
 import qualified Network.HTTP.Types.Status as HTTP
 import qualified Network.Wai.Utilities.Error as Wai
 import qualified Servant.Client as Servant
-import Wire.API.Federation.Client (FederationClientError (..))
+import Wire.API.Federation.Client (FederationClientError (..), FederationError (..))
 import qualified Wire.API.Federation.GRPC.Types as Proto
 
 data Error where

services/brig/src/Brig/API/Types.hs

@@ -43,7 +43,7 @@ import Brig.User.Auth.Cookie (RetryAfter (..))
 import Data.Id
 import Imports
 import qualified Network.Wai.Utilities.Error as Wai
-import Wire.API.Federation.Client (FederationClientError)
+import Wire.API.Federation.Client (FederationError)
 
 -------------------------------------------------------------------------------
 -- Successes
@@ -185,12 +185,6 @@ data SendLoginCodeError
   = SendLoginInvalidPhone Phone
   | SendLoginPasswordExists
 
-data FederationError
-  = FederationUnavailable Text
-  | FederationNotImplemented
-  | FederationNotConfigured
-  | FederationCallFailure FederationClientError
-
 data ClientError
   = ClientNotFound
   | ClientDataError !ClientDataError

services/brig/src/Brig/API/User.hs

@@ -151,6 +151,7 @@ import Imports
 import Network.Wai.Utilities
 import qualified System.Logger.Class as Log
 import System.Logger.Message
+import Wire.API.Federation.Client (FederationError (..))
 
 data AllowSCIMUpdates
   = AllowSCIMUpdates

services/brig/src/Brig/App.hs

@@ -131,6 +131,7 @@ import System.Logger.Class hiding (Settings, settings)
 import qualified System.Logger.Class as LC
 import qualified System.Logger.Extended as Log
 import Util.Options
+import Wire.API.Federation.Client (HasFederatorConfig (..))
 import Wire.API.User.Identity (Email)
 
 schemaVersion :: Int32
@@ -494,6 +495,10 @@ instance MonadUnliftIO m => MonadUnliftIO (AppT m) where
       withRunInIO $ \run ->
         inner (run . flip runReaderT r . unAppT)
 
+instance HasFederatorConfig AppIO where
+  federatorEndpoint = view federator
+  federationDomain = viewFederationDomain
+
 runAppT :: Env -> AppT m a -> m a
 runAppT e (AppT ma) = runReaderT ma e
 

services/brig/src/Brig/Federation/Client.hs

@@ -20,27 +20,20 @@
 
 module Brig.Federation.Client where
 
-import Brig.API.Types (FederationError (..))
-import Brig.App (AppIO, federator, viewFederationDomain)
+import Brig.App (AppIO)
 import Brig.Types (Prekey, PrekeyBundle)
 import qualified Brig.Types.Search as Public
 import Brig.Types.User
-import Control.Error.Util ((!?))
-import Control.Lens (view, (^.))
-import Control.Monad.Except (runExceptT)
-import Control.Monad.Trans.Except (ExceptT (..), throwE)
+import Control.Monad.Trans.Except (ExceptT (..))
 import Data.Domain
 import Data.Handle
 import Data.Id (ClientId, UserId)
 import Data.Qualified
 import qualified Data.Text as T
 import Imports
-import Mu.GRpc.Client.TyApps
 import qualified System.Logger.Class as Log
-import Util.Options (epHost, epPort)
 import Wire.API.Federation.API.Brig as FederatedBrig
-import Wire.API.Federation.Client (FederationClientError, FederatorClient, runFederatorClientWith)
-import Wire.API.Federation.GRPC.Client
+import Wire.API.Federation.Client (FederationError (..), executeFederated)
 import Wire.API.Message (UserClientMap, UserClients)
 import Wire.API.User.Client.Prekey (ClientPrekey)
 
@@ -86,23 +79,3 @@ searchUsers :: Domain -> SearchRequest -> FederationAppIO (Public.SearchResult P
 searchUsers domain searchTerm = do
   Log.warn $ Log.msg $ T.pack "Brig-federation: search call on remote backend"
   executeFederated domain $ FederatedBrig.searchUsers clientRoutes searchTerm
-
--- FUTUREWORK: It would be nice to share the client across all calls to
--- federator and not call this function on every invocation of federated
--- requests, but there are some issues in http2-client which might need some
--- fixing first. More context here:
--- https://github.com/lucasdicioccio/http2-client/issues/37
--- https://github.com/lucasdicioccio/http2-client/issues/49
-mkFederatorClient :: FederationAppIO GrpcClient
-mkFederatorClient = do
-  federatorEndpoint <- view federator !? FederationNotConfigured
-  let cfg = grpcClientConfigSimple (T.unpack (federatorEndpoint ^. epHost)) (fromIntegral (federatorEndpoint ^. epPort)) False
-  createGrpcClient cfg
-    >>= either (throwE . FederationUnavailable . reason) pure
-
-executeFederated :: Domain -> FederatorClient component (ExceptT FederationClientError FederationAppIO) a -> FederationAppIO a
-executeFederated targetDomain action = do
-  federatorClient <- mkFederatorClient
-  originDomain <- viewFederationDomain
-  runExceptT (runFederatorClientWith federatorClient targetDomain originDomain action)
-    >>= either (throwE . FederationCallFailure) pure

services/galley/galley.cabal

@@ -4,7 +4,7 @@ cabal-version: 1.12
 --
 -- see: https://github.com/sol/hpack
 --
--- hash: 8c007b9ab879e8b7a10f8066385a1dd36f86037c33233a58a441c680ac3831e7
+-- hash: 6b8bbc40705147b2cb4025609e4b8efbd3c91a97f426f13cc34530d11041a393
 
 name:           galley
 version:        0.83.0
@@ -105,6 +105,7 @@ library
     , http-client-openssl >=0.2
     , http-client-tls >=0.2.2
     , http-types >=0.8
+    , http2-client-grpc
     , imports
     , insert-ordered-containers
     , lens >=4.4
@@ -148,6 +149,7 @@ library
     , wai-utilities >=0.16
     , warp >=3.0
     , wire-api
+    , wire-api-federation
   default-language: Haskell2010
 
 executable galley
@@ -168,6 +170,7 @@ executable galley
     , ssl-util
     , types-common
     , wire-api
+    , wire-api-federation
   if flag(static)
     ld-options: -static
   default-language: Haskell2010
@@ -263,6 +266,7 @@ executable galley-integration
     , warp
     , warp-tls >=3.2
     , wire-api
+    , wire-api-federation
     , yaml
   default-language: Haskell2010
 
@@ -297,6 +301,7 @@ executable galley-migrate-data
     , types-common
     , unliftio
     , wire-api
+    , wire-api-federation
   if flag(static)
     ld-options: -static
   default-language: Haskell2010
@@ -349,6 +354,7 @@ executable galley-schema
     , text
     , tinylog
     , wire-api
+    , wire-api-federation
   if flag(static)
     ld-options: -static
   default-language: Haskell2010
@@ -384,4 +390,5 @@ test-suite galley-types-tests
     , wai
     , wai-predicates
     , wire-api
+    , wire-api-federation
   default-language: Haskell2010

services/galley/package.yaml

@@ -16,6 +16,7 @@ dependencies:
 - ssl-util
 - raw-strings-qq >=1.0
 - wire-api
+- wire-api-federation
 
 library:
   source-dirs: src
@@ -48,6 +49,7 @@ library:
   - http-client-openssl >=0.2
   - http-client-tls >=0.2.2
   - http-types >=0.8
+  - http2-client-grpc
   - insert-ordered-containers
   - lens >=4.4
   - metrics-wai >=0.4

services/galley/src/Galley/API/Util.hs

@@ -305,5 +305,5 @@ canDeleteMember deleter deletee
 --------------------------------------------------------------------------------
 -- Federation
 
-viewFederationDomain :: Galley Domain
+viewFederationDomain :: MonadReader Env m => m Domain
 viewFederationDomain = view (options . optSettings . setFederationDomain)

services/galley/src/Galley/App.hs

@@ -26,6 +26,7 @@ module Galley.App
     options,
     applog,
     manager,
+    federator,
     cstate,
     deleteQueue,
     createEnv,
@@ -88,6 +89,7 @@ import Ssl.Util
 import System.Logger.Class hiding (Error, info)
 import qualified System.Logger.Extended as Logger
 import Util.Options
+import Wire.API.Federation.Client (HasFederatorConfig (..))
 
 data DeleteItem = TeamItem TeamId UserId (Maybe ConnId)
   deriving (Eq, Ord, Show)
@@ -99,6 +101,7 @@ data Env = Env
     _options :: Opts,
     _applog :: Logger,
     _manager :: Manager,
+    _federator :: Maybe Endpoint, -- FUTUREWORK: should we use a better type here? E.g. to avoid fresh connections all the time?
     _cstate :: ClientState,
     _deleteQueue :: Q.Queue DeleteItem,
     _extEnv :: ExtEnv,
@@ -130,6 +133,10 @@ newtype Galley a = Galley
       MonadClient
     )
 
+instance HasFederatorConfig Galley where
+  federatorEndpoint = view federator
+  federationDomain = view (options . optSettings . setFederationDomain)
+
 fanoutLimit :: Galley (Range 1 Teams.HardTruncationLimit Int32)
 fanoutLimit = view options >>= return . currentFanoutLimit
 
@@ -186,7 +193,7 @@ createEnv m o = do
   l <- Logger.mkLogger (o ^. optLogLevel) (o ^. optLogNetStrings) (o ^. optLogFormat)
   mgr <- initHttpManager o
   validateOptions l o
-  Env def m o l mgr <$> initCassandra o l
+  Env def m o l mgr (o ^. optFederator) <$> initCassandra o l
     <*> Q.new 16000
     <*> initExtEnv
     <*> maybe (return Nothing) (fmap Just . Aws.mkEnv l mgr) (o ^. optJournal)