From 9aadc6ec0d7761e2025c6fab226a7792bbc0a129 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marko=20Dimja=C5=A1evi=C4=87?= Date: Wed, 18 Oct 2023 17:00:24 +0200 Subject: [PATCH] WIP: Refactoring fedQueueClient --- .../src/Wire/API/Federation/API.hs | 39 +++++-- .../src/Wire/API/Federation/API/Brig.hs | 2 +- .../API/Federation/API/Brig/Notifications.hs | 18 ++- .../src/Wire/API/Federation/API/Galley.hs | 2 +- .../Federation/API/Galley/Notifications.hs | 106 ++++++++++-------- .../API/Federation/BackendNotifications.hs | 85 +++++++------- .../src/Wire/API/Federation/Endpoint.hs | 4 +- .../API/Federation/HasNotificationEndpoint.hs | 35 ++++++ .../wire-api-federation.cabal | 2 + services/brig/src/Brig/Federation/Client.hs | 2 +- services/galley/src/Galley/API/Action.hs | 2 +- services/galley/src/Galley/API/Clients.hs | 2 +- services/galley/src/Galley/API/Internal.hs | 2 +- .../galley/src/Galley/API/MLS/Propagate.hs | 2 +- services/galley/src/Galley/API/Message.hs | 2 +- 15 files changed, 192 insertions(+), 113 deletions(-) create mode 100644 libs/wire-api-federation/src/Wire/API/Federation/HasNotificationEndpoint.hs diff --git a/libs/wire-api-federation/src/Wire/API/Federation/API.hs b/libs/wire-api-federation/src/Wire/API/Federation/API.hs index 5e6b294e122..da4dd979602 100644 --- a/libs/wire-api-federation/src/Wire/API/Federation/API.hs +++ b/libs/wire-api-federation/src/Wire/API/Federation/API.hs @@ -33,10 +33,16 @@ module Wire.API.Federation.API ) where +import Data.Aeson qualified as Aeson +import Data.ByteString.Char8 qualified as BS +import Data.Domain import Data.Kind import Data.Proxy +import Data.Text.Encoding qualified as TE import GHC.TypeLits import Imports +import Network.HTTP.Client qualified as HC +import Network.HTTP.Types import Servant import Servant.Client import Servant.Client.Core @@ -46,6 +52,7 @@ import Wire.API.Federation.API.Common import Wire.API.Federation.API.Galley import Wire.API.Federation.BackendNotifications import Wire.API.Federation.Client +import Wire.API.Federation.HasNotificationEndpoint import Wire.API.MakesFederatedCall import Wire.API.Routes.Named @@ -94,14 +101,32 @@ fedClient :: fedClient = clientIn (Proxy @api) (Proxy @m) fedQueueClient :: - forall (comp :: Component) (name :: Symbol) m api. - ( HasEmptyResponse api, - HasFedEndpoint comp api name, - HasClient m api, - m ~ FedQueueClient comp + forall tag api. + ( HasNotificationEndpoint tag, + -- api ~ NotificationAPI tag (NotificationComponent tag), + HasEmptyResponse api, + KnownSymbol (NotificationPath tag), + Aeson.ToJSON (Payload tag), + HasFedEndpoint (NotificationComponent tag) api (NotificationPath tag) ) => - Client m api -fedQueueClient = clientIn (Proxy @api) (Proxy @m) + Payload tag -> + FedQueueClient (NotificationComponent tag) () +fedQueueClient payload = do + -- TODO(md): instead of actually running the request here, enqueue a + -- 'BackendNotification' like it used to be in the RunClient instance for + -- FedQueueClient c. + env <- ask @FedQueueEnv + manager <- liftIO $ HC.newManager HC.defaultManagerSettings + let request = + HC.defaultRequest + { HC.method = methodPost, + HC.host = TE.encodeUtf8 . domainText $ env.targetDomain, + HC.path = BS.pack $ "/federation/" <> symbolVal (Proxy @(NotificationPath tag)), + HC.requestBody = HC.RequestBodyLBS . Aeson.encode $ payload, + HC.requestHeaders = [(hContentType, "application/json")] + } + + void . liftIO $ HC.httpLbs request manager fedClientIn :: forall (comp :: Component) (name :: Symbol) m api. diff --git a/libs/wire-api-federation/src/Wire/API/Federation/API/Brig.hs b/libs/wire-api-federation/src/Wire/API/Federation/API/Brig.hs index 929e687101c..8703e3d8501 100644 --- a/libs/wire-api-federation/src/Wire/API/Federation/API/Brig.hs +++ b/libs/wire-api-federation/src/Wire/API/Federation/API/Brig.hs @@ -77,7 +77,7 @@ type BrigApi = :<|> FedEndpoint "get-not-fully-connected-backends" DomainSet NonConnectedBackends -- All the notification endpoints that go through the queue-based -- federation client ('fedQueueClient'). - :<|> NotificationAPI + :<|> BrigNotificationAPI newtype DomainSet = DomainSet { domains :: Set Domain diff --git a/libs/wire-api-federation/src/Wire/API/Federation/API/Brig/Notifications.hs b/libs/wire-api-federation/src/Wire/API/Federation/API/Brig/Notifications.hs index 732d65641ed..efdc16722b9 100644 --- a/libs/wire-api-federation/src/Wire/API/Federation/API/Brig/Notifications.hs +++ b/libs/wire-api-federation/src/Wire/API/Federation/API/Brig/Notifications.hs @@ -21,7 +21,9 @@ import Data.Aeson import Data.Id import Data.Range import Imports +import Wire.API.Federation.Component import Wire.API.Federation.Endpoint +import Wire.API.Federation.HasNotificationEndpoint import Wire.API.Util.Aeson import Wire.Arbitrary @@ -37,6 +39,18 @@ data UserDeletedConnectionsNotification = UserDeletedConnectionsNotification deriving (Arbitrary) via (GenericUniform UserDeletedConnectionsNotification) deriving (FromJSON, ToJSON) via (CustomEncoded UserDeletedConnectionsNotification) +data BrigNotificationTag = OnUserDeletedConnectionsTag + deriving (Show, Eq, Generic, Bounded, Enum) + +instance HasNotificationEndpoint 'OnUserDeletedConnectionsTag where + type Payload 'OnUserDeletedConnectionsTag = UserDeletedConnectionsNotification + type NotificationPath 'OnUserDeletedConnectionsTag = "on-user-deleted-connections" + type NotificationComponent 'OnUserDeletedConnectionsTag = 'Brig + type + NotificationAPI 'OnUserDeletedConnectionsTag 'Brig = + NotificationFedEndpoint 'OnUserDeletedConnectionsTag + -- | All the notification endpoints return an 'EmptyResponse'. -type NotificationAPI = - NotificationFedEndpoint "on-user-deleted-connections" UserDeletedConnectionsNotification +type BrigNotificationAPI = + -- FUTUREWORK: Use NotificationAPI 'OnUserDeletedConnectionsTag 'Brig instead + NotificationFedEndpoint 'OnUserDeletedConnectionsTag diff --git a/libs/wire-api-federation/src/Wire/API/Federation/API/Galley.hs b/libs/wire-api-federation/src/Wire/API/Federation/API/Galley.hs index 48bccd4e4c1..f40417e303b 100644 --- a/libs/wire-api-federation/src/Wire/API/Federation/API/Galley.hs +++ b/libs/wire-api-federation/src/Wire/API/Federation/API/Galley.hs @@ -137,7 +137,7 @@ type GalleyApi = GetOne2OneConversationResponse -- All the notification endpoints that go through the queue-based -- federation client ('fedQueueClient'). - :<|> NotificationAPI + :<|> GalleyNotificationAPI data TypingDataUpdateRequest = TypingDataUpdateRequest { typingStatus :: TypingStatus, diff --git a/libs/wire-api-federation/src/Wire/API/Federation/API/Galley/Notifications.hs b/libs/wire-api-federation/src/Wire/API/Federation/API/Galley/Notifications.hs index b61e66fbc08..60e5c8efc71 100644 --- a/libs/wire-api-federation/src/Wire/API/Federation/API/Galley/Notifications.hs +++ b/libs/wire-api-federation/src/Wire/API/Federation/API/Galley/Notifications.hs @@ -24,7 +24,6 @@ import Data.Aeson import Data.Domain import Data.Id import Data.Json.Util -import Data.Kind import Data.List.NonEmpty import Data.Proxy import Data.Qualified @@ -37,6 +36,7 @@ import Servant.API import Wire.API.Conversation.Action import Wire.API.Federation.BackendNotifications import Wire.API.Federation.Endpoint +import Wire.API.Federation.HasNotificationEndpoint import Wire.API.MLS.SubConversation import Wire.API.MakesFederatedCall import Wire.API.Message @@ -52,65 +52,73 @@ data GalleyNotificationTag | OnUserDeletedConversationsTag deriving (Show, Eq, Generic, Bounded, Enum) -type family GalleyNotification (tag :: GalleyNotificationTag) :: Type where - GalleyNotification 'OnClientRemovedTag = ClientRemovedRequest - GalleyNotification 'OnMessageSentTag = RemoteMessage ConvId - GalleyNotification 'OnMLSMessageSentTag = RemoteMLSMessage - GalleyNotification 'OnConversationUpdatedTag = ConversationUpdate - GalleyNotification 'OnUserDeletedConversationsTag = UserDeletedConversationsNotification - --- | The central path component of a Galley notification endpoint -type family GNPath (tag :: GalleyNotificationTag) :: Symbol where - GNPath 'OnClientRemovedTag = "on-client-removed" - GNPath 'OnMessageSentTag = "on-message-sent" - GNPath 'OnMLSMessageSentTag = "on-mls-message-sent" - GNPath 'OnConversationUpdatedTag = "on-conversation-updated" - GNPath 'OnUserDeletedConversationsTag = "on-user-deleted-conversations" - -type GalleyNotifEndpoint (tag :: GalleyNotificationTag) = - NotificationFedEndpoint (GNPath tag) (GalleyNotification tag) - -type family GalleyNotificationToServantAPI (gn :: GalleyNotificationTag) :: Type where - GalleyNotificationToServantAPI 'OnClientRemovedTag = - NotificationFedEndpointWithMods - '[ MakesFederatedCall 'Galley "on-mls-message-sent" - ] - (GNPath 'OnClientRemovedTag) - (GalleyNotification 'OnClientRemovedTag) +instance HasNotificationEndpoint 'OnClientRemovedTag where + type Payload 'OnClientRemovedTag = ClientRemovedRequest + type NotificationPath 'OnClientRemovedTag = "on-client-removed" + type NotificationComponent 'OnClientRemovedTag = 'Galley + type + NotificationAPI 'OnClientRemovedTag 'Galley = + NotificationFedEndpointWithMods + '[ MakesFederatedCall 'Galley "on-mls-message-sent" + ] + (NotificationPath 'OnClientRemovedTag) + (Payload 'OnClientRemovedTag) + +instance HasNotificationEndpoint 'OnMessageSentTag where + type Payload 'OnMessageSentTag = RemoteMessage ConvId + type NotificationPath 'OnMessageSentTag = "on-message-sent" + type NotificationComponent 'OnMessageSentTag = 'Galley + -- used to notify this backend that a new message has been posted to a -- remote conversation - GalleyNotificationToServantAPI 'OnMessageSentTag = GalleyNotifEndpoint 'OnMessageSentTag - GalleyNotificationToServantAPI 'OnMLSMessageSentTag = GalleyNotifEndpoint 'OnMLSMessageSentTag + type NotificationAPI 'OnMessageSentTag 'Galley = NotificationFedEndpoint 'OnMessageSentTag + +instance HasNotificationEndpoint 'OnMLSMessageSentTag where + type Payload 'OnMLSMessageSentTag = RemoteMLSMessage + type NotificationPath 'OnMLSMessageSentTag = "on-mls-message-sent" + type NotificationComponent 'OnMLSMessageSentTag = 'Galley + type NotificationAPI 'OnMLSMessageSentTag 'Galley = NotificationFedEndpoint 'OnMLSMessageSentTag + +instance HasNotificationEndpoint 'OnConversationUpdatedTag where + type Payload 'OnConversationUpdatedTag = ConversationUpdate + type NotificationPath 'OnConversationUpdatedTag = "on-conversation-updated" + type NotificationComponent 'OnConversationUpdatedTag = 'Galley + -- used by the backend that owns a conversation to inform this backend of -- changes to the conversation - GalleyNotificationToServantAPI 'OnConversationUpdatedTag = - GalleyNotifEndpoint 'OnConversationUpdatedTag - GalleyNotificationToServantAPI 'OnUserDeletedConversationsTag = - NotificationFedEndpointWithMods - '[ MakesFederatedCall 'Galley "on-mls-message-sent", - MakesFederatedCall 'Galley "on-conversation-updated", - MakesFederatedCall 'Brig "api-version" - ] - (GNPath 'OnUserDeletedConversationsTag) - (GalleyNotification 'OnUserDeletedConversationsTag) + type NotificationAPI 'OnConversationUpdatedTag 'Galley = NotificationFedEndpoint 'OnConversationUpdatedTag + +instance HasNotificationEndpoint 'OnUserDeletedConversationsTag where + type Payload 'OnUserDeletedConversationsTag = UserDeletedConversationsNotification + type NotificationPath 'OnUserDeletedConversationsTag = "on-user-deleted-conversations" + type NotificationComponent 'OnUserDeletedConversationsTag = 'Galley + type + NotificationAPI 'OnUserDeletedConversationsTag 'Galley = + NotificationFedEndpointWithMods + '[ MakesFederatedCall 'Galley "on-mls-message-sent", + MakesFederatedCall 'Galley "on-conversation-updated", + MakesFederatedCall 'Brig "api-version" + ] + (NotificationPath 'OnUserDeletedConversationsTag) + (Payload 'OnUserDeletedConversationsTag) -- | All the notification endpoints return an 'EmptyResponse'. -type NotificationAPI = - GalleyNotificationToServantAPI 'OnClientRemovedTag - :<|> GalleyNotificationToServantAPI 'OnMessageSentTag - :<|> GalleyNotificationToServantAPI 'OnMLSMessageSentTag - :<|> GalleyNotificationToServantAPI 'OnConversationUpdatedTag - :<|> GalleyNotificationToServantAPI 'OnUserDeletedConversationsTag +type GalleyNotificationAPI = + NotificationAPI 'OnClientRemovedTag 'Galley + :<|> NotificationAPI 'OnMessageSentTag 'Galley + :<|> NotificationAPI 'OnMLSMessageSentTag 'Galley + :<|> NotificationAPI 'OnConversationUpdatedTag 'Galley + :<|> NotificationAPI 'OnUserDeletedConversationsTag 'Galley galleyToBackendNotification :: - forall tag. - KnownSymbol (GNPath tag) => - ToJSON (GalleyNotification tag) => + forall (tag :: GalleyNotificationTag). + KnownSymbol (NotificationPath tag) => + ToJSON (Payload tag) => Domain -> - GalleyNotification tag -> + Payload tag -> BackendNotification galleyToBackendNotification ownDomain gn = - let p = symbolVal (Proxy @(GNPath tag)) + let p = symbolVal (Proxy @(NotificationPath tag)) b = RawJson . encode $ gn in toNotif (T.pack . show $ p) b where diff --git a/libs/wire-api-federation/src/Wire/API/Federation/BackendNotifications.hs b/libs/wire-api-federation/src/Wire/API/Federation/BackendNotifications.hs index 3fa1aba2871..d13b0e6d359 100644 --- a/libs/wire-api-federation/src/Wire/API/Federation/BackendNotifications.hs +++ b/libs/wire-api-federation/src/Wire/API/Federation/BackendNotifications.hs @@ -6,22 +6,15 @@ module Wire.API.Federation.BackendNotifications where import Control.Exception import Control.Monad.Except import Data.Aeson -import Data.ByteString.Builder qualified as Builder -import Data.ByteString.Lazy qualified as LBS import Data.Domain import Data.Map qualified as Map -import Data.Sequence qualified as Seq import Data.Text qualified as Text -import Data.Text.Encoding import Data.Text.Lazy.Encoding qualified as TL import Imports import Network.AMQP qualified as Q import Network.AMQP.Types qualified as Q -import Network.HTTP.Types import Servant -import Servant.Client import Servant.Client.Core -import Servant.Types.SourceT import Wire.API.Federation.API.Common import Wire.API.Federation.Client import Wire.API.Federation.Component @@ -125,7 +118,7 @@ ensureQueue chan queue = do -- queue. Perhaps none of this should be servant code anymore. But it is here to -- allow smooth transition to RabbitMQ based notification pushing. -- --- Use 'Wire.API.Federation.API.fedQueueClient' to create and action and pass it +-- Use 'Wire.API.Federation.API.fedQueueClient' to create an action and pass it -- to 'enqueue' newtype FedQueueClient c a = FedQueueClient (ReaderT FedQueueEnv IO a) deriving (Functor, Applicative, Monad, MonadIO, MonadReader FedQueueEnv) @@ -142,41 +135,41 @@ data EnqueueError = EnqueueError String instance Exception EnqueueError -instance (KnownComponent c) => RunClient (FedQueueClient c) where - runRequestAcceptStatus :: Maybe [Status] -> Request -> FedQueueClient c Response - runRequestAcceptStatus _ req = do - env <- ask - bodyLBS <- case requestBody req of - Just (RequestBodyLBS lbs, _) -> pure lbs - Just (RequestBodyBS bs, _) -> pure (LBS.fromStrict bs) - Just (RequestBodySource src, _) -> liftIO $ do - errOrRes <- runExceptT $ runSourceT src - either (throwIO . EnqueueError) (pure . mconcat) errOrRes - Nothing -> pure mempty - let notif = - BackendNotification - { ownDomain = env.originDomain, - targetComponent = componentVal @c, - path = decodeUtf8 $ LBS.toStrict $ Builder.toLazyByteString req.requestPath, - body = RawJson bodyLBS - } - let msg = - Q.newMsg - { Q.msgBody = encode notif, - Q.msgDeliveryMode = Just (env.deliveryMode), - Q.msgContentType = Just "application/json" - } - -- Empty string means default exchange - exchange = "" - liftIO $ do - ensureQueue env.channel env.targetDomain._domainText - void $ Q.publishMsg env.channel exchange (routingKey env.targetDomain._domainText) msg - pure $ - Response - { responseHttpVersion = http20, - responseStatusCode = status200, - responseHeaders = Seq.singleton (hContentType, "application/json"), - responseBody = "{}" - } - throwClientError :: ClientError -> FedQueueClient c a - throwClientError = liftIO . throwIO +-- instance (KnownComponent c) => RunClient (FedQueueClient c) where +-- runRequestAcceptStatus :: Maybe [Status] -> Request -> FedQueueClient c Response +-- runRequestAcceptStatus _ req = do +-- env <- ask +-- bodyLBS <- case requestBody req of +-- Just (RequestBodyLBS lbs, _) -> pure lbs +-- Just (RequestBodyBS bs, _) -> pure (LBS.fromStrict bs) +-- Just (RequestBodySource src, _) -> liftIO $ do +-- errOrRes <- runExceptT $ runSourceT src +-- either (throwIO . EnqueueError) (pure . mconcat) errOrRes +-- Nothing -> pure mempty +-- let notif = +-- BackendNotification +-- { ownDomain = env.originDomain, +-- targetComponent = componentVal @c, +-- path = decodeUtf8 $ LBS.toStrict $ Builder.toLazyByteString req.requestPath, +-- body = RawJson bodyLBS +-- } +-- let msg = +-- Q.newMsg +-- { Q.msgBody = encode notif, +-- Q.msgDeliveryMode = Just (env.deliveryMode), +-- Q.msgContentType = Just "application/json" +-- } +-- -- Empty string means default exchange +-- exchange = "" +-- liftIO $ do +-- ensureQueue env.channel env.targetDomain._domainText +-- void $ Q.publishMsg env.channel exchange (routingKey env.targetDomain._domainText) msg +-- pure $ +-- Response +-- { responseHttpVersion = http20, +-- responseStatusCode = status200, +-- responseHeaders = Seq.singleton (hContentType, "application/json"), +-- responseBody = "{}" +-- } +-- throwClientError :: ClientError -> FedQueueClient c a +-- throwClientError = liftIO . throwIO diff --git a/libs/wire-api-federation/src/Wire/API/Federation/Endpoint.hs b/libs/wire-api-federation/src/Wire/API/Federation/Endpoint.hs index 323f161c899..664835848f0 100644 --- a/libs/wire-api-federation/src/Wire/API/Federation/Endpoint.hs +++ b/libs/wire-api-federation/src/Wire/API/Federation/Endpoint.hs @@ -26,6 +26,7 @@ import Servant.API import Wire.API.ApplyMods import Wire.API.Federation.API.Common import Wire.API.Federation.Domain +import Wire.API.Federation.HasNotificationEndpoint import Wire.API.Routes.Named type FedEndpointWithMods (mods :: [Type]) name input output = @@ -41,7 +42,8 @@ type NotificationFedEndpointWithMods (mods :: [Type]) name input = type FedEndpoint name input output = FedEndpointWithMods '[] name input output -type NotificationFedEndpoint name input = FedEndpoint name input EmptyResponse +type NotificationFedEndpoint tag = + FedEndpoint (NotificationPath tag) (Payload tag) EmptyResponse type StreamingFedEndpoint name input output = Named diff --git a/libs/wire-api-federation/src/Wire/API/Federation/HasNotificationEndpoint.hs b/libs/wire-api-federation/src/Wire/API/Federation/HasNotificationEndpoint.hs new file mode 100644 index 00000000000..d47828b65fc --- /dev/null +++ b/libs/wire-api-federation/src/Wire/API/Federation/HasNotificationEndpoint.hs @@ -0,0 +1,35 @@ +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2023 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.API.Federation.HasNotificationEndpoint where + +import Data.Kind +import GHC.TypeLits +import Wire.API.Federation.Component + +class HasNotificationEndpoint t where + -- | The type of the payload for this endpoint + type Payload t :: Type + + -- | The central path component of a notification endpoint + type NotificationPath t :: Symbol + + -- | The server component this endpoint is associated with + type NotificationComponent t :: Component + + -- | The Servant API endpoint type + type NotificationAPI t (c :: Component) :: Type diff --git a/libs/wire-api-federation/wire-api-federation.cabal b/libs/wire-api-federation/wire-api-federation.cabal index 7a3a0228e79..3eded9da156 100644 --- a/libs/wire-api-federation/wire-api-federation.cabal +++ b/libs/wire-api-federation/wire-api-federation.cabal @@ -29,6 +29,7 @@ library Wire.API.Federation.Domain Wire.API.Federation.Endpoint Wire.API.Federation.Error + Wire.API.Federation.HasNotificationEndpoint Wire.API.Federation.Version other-modules: Paths_wire_api_federation @@ -91,6 +92,7 @@ library , containers , exceptions , HsOpenSSL + , http-client , http-media , http-types , http2 diff --git a/services/brig/src/Brig/Federation/Client.hs b/services/brig/src/Brig/Federation/Client.hs index f0068f64320..87c44ec4465 100644 --- a/services/brig/src/Brig/Federation/Client.hs +++ b/services/brig/src/Brig/Federation/Client.hs @@ -152,7 +152,7 @@ notifyUserDeleted self remotes = do Just chanVar -> do enqueueNotification (tDomain self) remoteDomain Q.Persistent chanVar $ void $ - fedQueueClient @'Brig @"on-user-deleted-connections" notif + fedQueueClient @'OnUserDeletedConnectionsTag notif Nothing -> Log.err $ Log.msg ("Federation error while notifying remote backends of a user deletion." :: ByteString) diff --git a/services/galley/src/Galley/API/Action.hs b/services/galley/src/Galley/API/Action.hs index e6994075680..d8d5c530cdf 100644 --- a/services/galley/src/Galley/API/Action.hs +++ b/services/galley/src/Galley/API/Action.hs @@ -883,7 +883,7 @@ notifyConversationAction tag quid notifyOrigDomain con lconv targets action = do -- 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 + then fedQueueClient @'OnConversationUpdatedTag update $> Nothing else pure (Just update) -- notify local participants and bots diff --git a/services/galley/src/Galley/API/Clients.hs b/services/galley/src/Galley/API/Clients.hs index b264724a043..044447c488d 100644 --- a/services/galley/src/Galley/API/Clients.hs +++ b/services/galley/src/Galley/API/Clients.hs @@ -137,5 +137,5 @@ rmClientH (usr ::: cid) = do removeRemoteMLSClients :: Range 1 1000 [Remote ConvId] -> Sem r () removeRemoteMLSClients convIds = do for_ (bucketRemote (fromRange convIds)) $ \remoteConvs -> - let rpc = void $ fedQueueClient @'Galley @"on-client-removed" (ClientRemovedRequest usr cid (tUnqualified remoteConvs)) + let rpc = void $ fedQueueClient @'OnClientRemovedTag (ClientRemovedRequest usr cid (tUnqualified remoteConvs)) in enqueueNotification remoteConvs Q.Persistent rpc diff --git a/services/galley/src/Galley/API/Internal.hs b/services/galley/src/Galley/API/Internal.hs index 2830fc16d43..b33bab98ac5 100644 --- a/services/galley/src/Galley/API/Internal.hs +++ b/services/galley/src/Galley/API/Internal.hs @@ -420,7 +420,7 @@ rmUser lusr conn = do leaveRemoteConversations cids = for_ (bucketRemote (fromRange cids)) $ \remoteConvs -> do let userDelete = UserDeletedConversationsNotification (tUnqualified lusr) (unsafeRange (tUnqualified remoteConvs)) - let rpc = void $ fedQueueClient @'Galley @"on-user-deleted-conversations" userDelete + let rpc = void $ fedQueueClient @'OnUserDeletedConversationsTag userDelete enqueueNotification remoteConvs Q.Persistent rpc -- FUTUREWORK: Add a retry mechanism if there are federation errrors. diff --git a/services/galley/src/Galley/API/MLS/Propagate.hs b/services/galley/src/Galley/API/MLS/Propagate.hs index e9f6ac089d7..6b17a3a8a62 100644 --- a/services/galley/src/Galley/API/MLS/Propagate.hs +++ b/services/galley/src/Galley/API/MLS/Propagate.hs @@ -89,7 +89,7 @@ propagateMessage qusr mSenderClient lConvOrSub con msg cm = do -- send to remotes (either (logRemoteNotificationError @"on-mls-message-sent") (const (pure ())) <=< enqueueNotificationsConcurrently Q.Persistent (map remoteMemberQualify rmems)) $ \rs -> - fedQueueClient @'Galley @"on-mls-message-sent" $ + fedQueueClient @'OnMLSMessageSentTag $ RemoteMLSMessage { time = now, sender = qusr, diff --git a/services/galley/src/Galley/API/Message.hs b/services/galley/src/Galley/API/Message.hs index a78166d3e3a..66657736a6c 100644 --- a/services/galley/src/Galley/API/Message.hs +++ b/services/galley/src/Galley/API/Message.hs @@ -663,7 +663,7 @@ sendRemoteMessages domain now sender senderClient lcnv metadata messages = (hand transient = mmTransient metadata, recipients = UserClientMap rcpts } - let rpc = void $ fedQueueClient @'Galley @"on-message-sent" rm + let rpc = void $ fedQueueClient @'OnMessageSentTag rm enqueueNotification domain Q.Persistent rpc where handle :: Either FederationError a -> Sem r (Set (UserId, ClientId))