galley: Send on-user-deleted-conversations backend notification through RabbitMQ

changelog.d/0-release-notes/background-worker

@@ -1,3 +1,5 @@
+Introduce background-worker (#3276, #3314, #3333)
+
 This release introduces a new component: background-worker. This is currently
 only used to forward notifications to federated backends. Enabling federation in
 the wire-server helm chart automatically installs this component.
@@ -18,6 +20,16 @@ brig:
     rabbitmq:
       username: <YOUR_USERNAME>
       password: <YOUR_PASSWORD>
+galley:
+  config:
+    rabbitmq:
+      host: rabbitmq
+      port: 5672
+      vHost: /
+  secrets:
+    rabbitmq:
+      username: <YOUR_USERNAME>
+      password: <YOUR_PASSWORD>
 background-worker:
   config:
     rabbitmq:

charts/galley/templates/configmap.yaml

@@ -38,6 +38,8 @@ data:
     federator:
       host: federator
       port: 8080
+    rabbitmq:
+{{ toYaml .rabbitmq | indent 6}}
     {{- end }}
 
     {{- if (.journal) }}

charts/galley/templates/deployment.yaml

@@ -79,7 +79,19 @@ spec:
           - name: NO_PROXY
             value: {{ join "," .noProxyList | quote }}
           {{- end }}
-          {{- end }} 
+          {{- end }}
+          {{- if .Values.config.enableFederation }}
+          - name: RABBITMQ_USERNAME
+            valueFrom:
+              secretKeyRef:
+                name: galley
+                key: rabbitmqUsername
+          - name: RABBITMQ_PASSWORD
+            valueFrom:
+              secretKeyRef:
+                name: galley
+                key: rabbitmqPassword
+          {{- end }}
           ports:
             - containerPort: {{ .Values.service.internalPort }}
           livenessProbe:

charts/galley/templates/secret.yaml

@@ -13,4 +13,8 @@ data:
   {{- if .Values.secrets.mlsPrivateKeys.removal.ed25519 }}
   removal_ed25519.pem: {{ .Values.secrets.mlsPrivateKeys.removal.ed25519 | b64enc | quote }}
   {{- end -}}
+  {{- if $.Values.config.enableFederation }}
+  rabbitmqUsername: {{ .Values.secrets.rabbitmq.username | b64enc | quote }}
+  rabbitmqPassword: {{ .Values.secrets.rabbitmq.password | b64enc | quote }}
+  {{- end }}
   {{- end -}}

charts/galley/templates/tests/galley-integration.yaml

@@ -61,6 +61,12 @@ spec:
       value: "dummy"
     - name: AWS_REGION
       value: "eu-west-1"
+    {{- if .Values.config.enableFederation }}
+    - name: RABBITMQ_USERNAME
+      value: "guest"
+    - name: RABBITMQ_PASSWORD
+      value: "guest"
+    {{- end }}
     resources:
       requests:
         memory: "512Mi"

charts/galley/values.yaml

@@ -23,6 +23,11 @@ config:
     host: aws-cassandra
     replicaCount: 3
   enableFederation: false # keep enableFederation default in sync with brig and cargohold chart's config.enableFederation as well as wire-server chart's tags.federation
+  # Not used if enableFederation is false
+  rabbitmq:
+    host: rabbitmq
+    port: 5672
+    vHost: /
   settings:
     httpPoolSize: 128
     maxTeamSize: 10000

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

@@ -203,6 +203,9 @@ galley:
           -----BEGIN PRIVATE KEY-----
           MC4CAQAwBQYDK2VwBCIEIAocCDXsKIAjb65gOUn5vEF0RIKnVJkKR4ebQzuZ709c
           -----END PRIVATE KEY-----
+    rabbitmq:
+      username: {{ .Values.rabbitmqUsername }}
+      password: {{ .Values.rabbitmqPassword }}
 
 gundeck:
   replicaCount: 1

libs/wire-api/src/Wire/API/Routes/Internal/Galley.hs

@@ -184,7 +184,6 @@ type InternalAPIBase =
            ( Summary
                "Remove a user from their teams and conversations and erase their clients"
                :> MakesFederatedCall 'Galley "on-conversation-updated"
-               :> MakesFederatedCall 'Galley "on-user-deleted-conversations"
                :> MakesFederatedCall 'Galley "on-mls-message-sent"
                :> ZLocalUser
                :> ZOptConn

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

@@ -299,7 +299,6 @@ type SelfAPI =
           :> CanThrow 'MissingAuth
           :> CanThrow 'DeleteCodePending
           :> CanThrow 'OwnerDeletingSelf
-          :> MakesFederatedCall 'Brig "on-user-deleted-connections"
           :> ZUser
           :> "self"
           :> ReqBody '[JSON] DeleteUser
@@ -311,7 +310,6 @@ type SelfAPI =
     Named
       "put-self"
       ( Summary "Update your profile."
-          :> MakesFederatedCall 'Brig "on-user-deleted-connections"
           :> ZUser
           :> ZConn
           :> "self"
@@ -337,7 +335,6 @@ type SelfAPI =
           :> Description
                "Your phone number can only be removed if you also have an \
                \email address and a password."
-          :> MakesFederatedCall 'Brig "on-user-deleted-connections"
           :> ZUser
           :> ZConn
           :> "self"
@@ -353,7 +350,6 @@ type SelfAPI =
           :> Description
                "Your email address can only be removed if you also have a \
                \phone number."
-          :> MakesFederatedCall 'Brig "on-user-deleted-connections"
           :> ZUser
           :> ZConn
           :> "self"
@@ -386,7 +382,6 @@ type SelfAPI =
     :<|> Named
            "change-locale"
            ( Summary "Change your locale."
-               :> MakesFederatedCall 'Brig "on-user-deleted-connections"
                :> ZUser
                :> ZConn
                :> "self"
@@ -397,7 +392,6 @@ type SelfAPI =
     :<|> Named
            "change-handle"
            ( Summary "Change your handle."
-               :> MakesFederatedCall 'Brig "on-user-deleted-connections"
                :> ZUser
                :> ZConn
                :> "self"
@@ -449,7 +443,6 @@ type AccountAPI =
              "If the environment where the registration takes \
              \place is private and a registered email address or phone \
              \number is not whitelisted, a 403 error is returned."
-        :> MakesFederatedCall 'Brig "on-user-deleted-connections"
         :> "register"
         :> ReqBody '[JSON] NewUserPublic
         :> MultiVerb 'POST '[JSON] RegisterResponses (Either RegisterError RegisterSuccess)
@@ -460,7 +453,6 @@ type AccountAPI =
     :<|> Named
            "verify-delete"
            ( Summary "Verify account deletion with a code."
-               :> MakesFederatedCall 'Brig "on-user-deleted-connections"
                :> CanThrow 'InvalidCode
                :> "delete"
                :> ReqBody '[JSON] VerifyDeleteUser
@@ -473,7 +465,6 @@ type AccountAPI =
            "get-activate"
            ( Summary "Activate (i.e. confirm) an email address or phone number."
                :> Description "See also 'POST /activate' which has a larger feature set."
-               :> MakesFederatedCall 'Brig "on-user-deleted-connections"
                :> CanThrow 'UserKeyExists
                :> CanThrow 'InvalidActivationCodeWrongUser
                :> CanThrow 'InvalidActivationCodeWrongCode
@@ -499,7 +490,6 @@ type AccountAPI =
                :> Description
                     "Activation only succeeds once and the number of \
                     \failed attempts for a valid key is limited."
-               :> MakesFederatedCall 'Brig "on-user-deleted-connections"
                :> CanThrow 'UserKeyExists
                :> CanThrow 'InvalidActivationCodeWrongUser
                :> CanThrow 'InvalidActivationCodeWrongCode
@@ -704,7 +694,6 @@ type UserClientAPI =
   Named
     "add-client"
     ( Summary "Register a new client"
-        :> MakesFederatedCall 'Brig "on-user-deleted-connections"
         :> CanThrow 'TooManyClients
         :> CanThrow 'MissingAuth
         :> CanThrow 'MalformedPrekeys
@@ -1261,7 +1250,6 @@ type AuthAPI =
              \ Every other combination is invalid.\
              \ Access tokens can be given as query parameter or authorisation\
              \ header, with the latter being preferred."
-        :> MakesFederatedCall 'Brig "on-user-deleted-connections"
         :> QueryParam "client_id" ClientId
         :> Cookies '["zuid" ::: SomeUserToken]
         :> Bearer SomeAccessToken
@@ -1292,7 +1280,6 @@ type AuthAPI =
            ( "login"
                :> Summary "Authenticate a user to obtain a cookie and first access token"
                :> Description "Logins are throttled at the server's discretion"
-               :> MakesFederatedCall 'Brig "on-user-deleted-connections"
                :> ReqBody '[JSON] Login
                :> QueryParam'
                     [ Optional,

services/brig/src/Brig/App.hs

@@ -203,8 +203,16 @@ data Env = Env
 
 makeLenses ''Env
 
+validateOptions :: Opts -> IO ()
+validateOptions o =
+  case (o.federatorInternal, o.rabbitmq) of
+    (Nothing, Just _) -> error "RabbitMQ config is specified and federator is not, please specify both or none"
+    (Just _, Nothing) -> error "Federator is specified and RabbitMQ config is not, please specify both or none"
+    _ -> pure ()
+
 newEnv :: Opts -> IO Env
 newEnv o = do
+  validateOptions o
   Just md5 <- getDigestByName "MD5"
   Just sha256 <- getDigestByName "SHA256"
   Just sha512 <- getDigestByName "SHA512"

services/galley/default.nix

@@ -7,6 +7,7 @@
 , aeson-qq
 , amazonka
 , amazonka-sqs
+, amqp
 , asn1-encoding
 , asn1-types
 , async
@@ -135,6 +136,7 @@ mkDerivation {
     aeson
     amazonka
     amazonka-sqs
+    amqp
     asn1-encoding
     asn1-types
     async

services/galley/galley.cabal

@@ -92,6 +92,7 @@ library
     Galley.Data.TeamNotifications
     Galley.Data.Types
     Galley.Effects
+    Galley.Effects.BackendNotificationQueueAccess
     Galley.Effects.BotAccess
     Galley.Effects.BrigAccess
     Galley.Effects.ClientStore
@@ -121,6 +122,7 @@ library
     Galley.External.LegalHoldService
     Galley.External.LegalHoldService.Internal
     Galley.External.LegalHoldService.Types
+    Galley.Intra.BackendNotificationQueue
     Galley.Intra.Client
     Galley.Intra.Effects
     Galley.Intra.Federator
@@ -195,6 +197,7 @@ library
       aeson                  >=2.0.1.0
     , amazonka               >=1.4.5
     , amazonka-sqs           >=1.4.5
+    , amqp
     , asn1-encoding
     , asn1-types
     , async                  >=2.0

services/galley/galley.integration.yaml

@@ -25,6 +25,11 @@ federator:
   host: 127.0.0.1
   port: 8097
 
+rabbitmq:
+  host: 127.0.0.1
+  port: 5672
+  vHost: /
+
 settings:
   httpPoolSize: 128
   maxTeamSize: 32

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

@@ -51,6 +51,7 @@ import Galley.API.Util
 import Galley.App
 import qualified Galley.Data.Conversation as Data
 import Galley.Effects
+import Galley.Effects.BackendNotificationQueueAccess
 import Galley.Effects.ClientStore
 import Galley.Effects.ConversationStore
 import Galley.Effects.FederatorAccess
@@ -68,6 +69,7 @@ import Galley.Types.Bot.Service
 import Galley.Types.Conversations.Members (RemoteMember (rmId))
 import Galley.Types.UserList
 import Imports hiding (head)
+import qualified Network.AMQP as Q
 import Network.Wai.Predicate hiding (Error, err)
 import qualified Network.Wai.Predicate as Predicate
 import Network.Wai.Routing hiding (App, route, toList)
@@ -301,7 +303,8 @@ rmUser ::
   forall p1 p2 r.
   ( p1 ~ CassandraPaging,
     p2 ~ InternalPaging,
-    ( Member BrigAccess r,
+    ( Member BackendNotificationQueueAccess r,
+      Member BrigAccess r,
       Member ClientStore r,
       Member ConversationStore r,
       Member (Error InternalError) r,
@@ -401,12 +404,11 @@ rmUser lusr conn = do
         >>= logAndIgnoreError "Error in onConversationUpdated call" (qUnqualified qUser)
 
     leaveRemoteConversations :: Range 1 UserDeletedNotificationMaxConvs [Remote ConvId] -> Sem r ()
-    leaveRemoteConversations cids = do
+    leaveRemoteConversations cids =
       for_ (bucketRemote (fromRange cids)) $ \remoteConvs -> do
         let userDelete = UserDeletedConversationsNotification (tUnqualified lusr) (unsafeRange (tUnqualified remoteConvs))
-        let rpc = fedClient @'Galley @"on-user-deleted-conversations" userDelete
-        runFederatedEither remoteConvs rpc
-          >>= logAndIgnoreError "Error in onUserDeleted call" (tUnqualified lusr)
+        let rpc = void $ fedQueueClient @'Galley @"on-user-deleted-conversations" userDelete
+        enqueueNotification remoteConvs Q.Persistent rpc
 
     -- FUTUREWORK: Add a retry mechanism if there are federation errrors.
     -- See https://wearezeta.atlassian.net/browse/SQCORE-1091

services/galley/src/Galley/App.hs

@@ -78,6 +78,7 @@ import Galley.Effects.FireAndForget (interpretFireAndForget)
 import Galley.Effects.WaiRoutes.IO
 import Galley.Env
 import Galley.External
+import Galley.Intra.BackendNotificationQueue
 import Galley.Intra.Effects
 import Galley.Intra.Federator
 import Galley.Keys
@@ -148,6 +149,10 @@ validateOptions l o = do
     error "setMaxConvSize cannot be > setTruncationLimit"
   when (settings ^. setMaxTeamSize < optFanoutLimit) $
     error "setMaxTeamSize cannot be < setTruncationLimit"
+  case (o ^. optFederator, o ^. optRabbitmq) of
+    (Nothing, Just _) -> error "RabbitMQ config is specified and federator is not, please specify both or none"
+    (Just _, Nothing) -> error "Federator is specified and RabbitMQ config is not, please specify both or none"
+    _ -> pure ()
 
 createEnv :: Metrics -> Opts -> IO Env
 createEnv m o = do
@@ -161,6 +166,7 @@ createEnv m o = do
     <*> initExtEnv
     <*> maybe (pure Nothing) (fmap Just . Aws.mkEnv l mgr) (o ^. optJournal)
     <*> loadAllMLSKeys (fold (o ^. optSettings . setMlsPrivateKeyPaths))
+    <*> mkRabbitMqChannel l o
 
 initCassandra :: Opts -> Logger -> IO ClientState
 initCassandra o l = do
@@ -277,6 +283,7 @@ evalGalley e =
     . interpretClientStoreToCassandra
     . interpretFireAndForget
     . interpretBotAccess
+    . interpretBackendNotificationQueueAccess
     . interpretFederatorAccess
     . interpretExternalAccess
     . interpretGundeckAccess

services/galley/src/Galley/Effects.hs

@@ -62,6 +62,7 @@ where
 import Data.Id
 import Data.Qualified
 import Data.Time.Clock
+import Galley.Effects.BackendNotificationQueueAccess
 import Galley.Effects.BotAccess
 import Galley.Effects.BrigAccess
 import Galley.Effects.ClientStore
@@ -101,6 +102,7 @@ type GalleyEffects1 =
      GundeckAccess,
      ExternalAccess,
      FederatorAccess,
+     BackendNotificationQueueAccess,
      BotAccess,
      FireAndForget,
      ClientStore,

services/galley/src/Galley/Effects/BackendNotificationQueueAccess.hs

@@ -0,0 +1,21 @@
+{-# LANGUAGE TemplateHaskell #-}
+
+module Galley.Effects.BackendNotificationQueueAccess where
+
+import Data.Qualified
+import Imports
+import qualified Network.AMQP as Q
+import Polysemy
+import Wire.API.Federation.BackendNotifications
+import Wire.API.Federation.Component
+import Wire.API.Federation.Error
+
+data BackendNotificationQueueAccess m a where
+  EnqueueNotification ::
+    KnownComponent c =>
+    Remote x ->
+    Q.DeliveryMode ->
+    FedQueueClient c () ->
+    BackendNotificationQueueAccess m (Either FederationError ())
+
+makeSem ''BackendNotificationQueueAccess