[WPB-3867] Use queueing federation client for a federation API endpoint

changelog.d/6-federation/wpb-3867-queue-endpoints

@@ -0,0 +1 @@
+Constrain which federation endpoints can be used via the queueing federation client

changelog.d/6-federation/wpb-3867-unreachable-users

@@ -0,0 +1 @@
+There is a breaking change in the "on-mls-message-sent" federation endpoint due to queueing. Now that there is retrying because of queueing, the endpoint can no longer respond with a list of unreachable users.

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

@@ -37,10 +37,12 @@ import Data.Kind
 import Data.Proxy
 import GHC.TypeLits
 import Imports
+import Servant
 import Servant.Client
 import Servant.Client.Core
 import Wire.API.Federation.API.Brig
 import Wire.API.Federation.API.Cargohold
+import Wire.API.Federation.API.Common
 import Wire.API.Federation.API.Galley
 import Wire.API.Federation.BackendNotifications
 import Wire.API.Federation.Client
@@ -64,6 +66,20 @@ type HasFedEndpoint comp api name = (HasUnsafeFedEndpoint comp api name)
 -- you to forget about some federated calls.
 type HasUnsafeFedEndpoint comp api name = 'Just api ~ LookupEndpoint (FedApi comp) name
 
+-- | Constrains which endpoints can be used with FedQueueClient.
+--
+-- Since the servant client implementation underlying FedQueueClient is
+-- returning a "fake" response consisting of an empty object, we need to make
+-- sure that an API type is compatible with an empty response if we want to
+-- invoke it using `fedQueueClient`
+class HasEmptyResponse api
+
+instance HasEmptyResponse (Post '[JSON] EmptyResponse)
+
+instance HasEmptyResponse api => HasEmptyResponse (x :> api)
+
+instance HasEmptyResponse api => HasEmptyResponse (Named name api)
+
 -- | Return a client for a named endpoint.
 --
 -- This function introduces an 'AddAnnotation' constraint, which is
@@ -79,7 +95,11 @@ fedClient = clientIn (Proxy @api) (Proxy @m)
 
 fedQueueClient ::
   forall (comp :: Component) (name :: Symbol) m api.
-  (HasFedEndpoint comp api name, HasClient m api, m ~ FedQueueClient comp) =>
+  ( HasEmptyResponse api,
+    HasFedEndpoint comp api name,
+    HasClient m api,
+    m ~ FedQueueClient comp
+  ) =>
   Client m api
 fedQueueClient = clientIn (Proxy @api) (Proxy @m)
 

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

@@ -40,7 +40,6 @@ import Wire.API.MLS.SubConversation
 import Wire.API.MakesFederatedCall
 import Wire.API.Message
 import Wire.API.Routes.Public.Galley.Messaging
-import Wire.API.Unreachable
 import Wire.API.Util.Aeson (CustomEncoded (..), CustomEncodedLensable (..))
 import Wire.Arbitrary (Arbitrary, GenericUniform (..))
 
@@ -97,7 +96,7 @@ type GalleyApi =
            ConversationUpdateRequest
            ConversationUpdateResponse
     :<|> FedEndpoint "mls-welcome" MLSWelcomeRequest MLSWelcomeResponse
-    :<|> FedEndpoint "on-mls-message-sent" RemoteMLSMessage RemoteMLSMessageResponse
+    :<|> FedEndpoint "on-mls-message-sent" RemoteMLSMessage EmptyResponse
     :<|> FedEndpointWithMods
            '[ MakesFederatedCall 'Galley "on-conversation-updated",
               MakesFederatedCall 'Galley "on-mls-message-sent",
@@ -419,7 +418,7 @@ data MLSMessageResponse
     MLSMessageResponseUnreachableBackends (Set Domain)
   | -- | If the list of unreachable users is non-empty, it corresponds to users
     -- that an application message could not be sent to.
-    MLSMessageResponseUpdates [ConversationUpdate] (Maybe UnreachableUsers)
+    MLSMessageResponseUpdates [ConversationUpdate]
   | MLSMessageResponseNonFederatingBackends NonFederatingBackends
   deriving stock (Eq, Show, Generic)
   deriving (ToJSON, FromJSON) via (CustomEncoded MLSMessageResponse)

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

@@ -786,47 +786,36 @@ notifyConversationAction tag quid notifyOrigDomain con lconv targets action = do
   now <- input
   let lcnv = fmap (.convId) lconv
       e = conversationActionToEvent tag now quid (tUntagged lcnv) Nothing action
-
-  let mkUpdate uids =
+      mkUpdate uids =
         ConversationUpdate
           now
           quid
           (tUnqualified lcnv)
           uids
           (SomeConversationAction tag action)
-
-  update <- do
-    let remoteTargets = toList (bmRemotes targets)
-    updates <-
-      enqueueNotificationsConcurrently Q.Persistent remoteTargets $ \ruids -> do
+      handleError :: FederationError -> Sem r (Maybe ConversationUpdate)
+      handleError fedErr =
+        logRemoteNotificationError @"on-conversation-updated" fedErr $> Nothing
+
+  update <-
+    fmap (fromMaybe (mkUpdate []))
+      . (either handleError (pure . asum . map tUnqualified))
+      <=< enqueueNotificationsConcurrently Q.Persistent (toList (bmRemotes targets))
+      $ \ruids -> do
         let update = mkUpdate (tUnqualified ruids)
         -- if notifyOrigDomain is false, filter out user from quid's domain,
         -- because quid's backend will update local state and notify its users
         -- itself using the ConversationUpdate returned by this function
         if notifyOrigDomain || tDomain ruids /= qDomain quid
           then fedQueueClient @'Galley @"on-conversation-updated" update $> Nothing
           else pure (Just update)
-    case partitionEithers updates of
-      (ls :: [Remote ([UserId], FederationError)], rs) -> do
-        for_ ls $
-          logError
-            "on-conversation-updated"
-            "An error occurred while communicating with federated server: "
-        pure $ fromMaybe (mkUpdate []) . asum . map tUnqualified $ rs
 
   -- notify local participants and bots
   pushConversationEvent con e (qualifyAs lcnv (bmLocals targets)) (bmBots targets)
 
   -- return both the event and the 'ConversationUpdate' structure corresponding
   -- to the originating domain (if it is remote)
   pure $ LocalConversationUpdate e update
-  where
-    logError :: String -> String -> Remote (a, FederationError) -> Sem r ()
-    logError field msg e =
-      P.warn $
-        Log.field "federation call" field
-          . Log.field "domain" (_domainText (tDomain e))
-          . Log.msg (msg <> displayException (snd (tUnqualified e)))
 
 -- | Update the local database with information on conversation members joining
 -- or leaving. Finally, push out notifications to local users.