[WPB-10308] Use RabbitMQ queues for notifications

Makefile

@@ -49,9 +49,14 @@ install: init
 	./hack/bin/cabal-run-all-tests.sh
 	./hack/bin/cabal-install-artefacts.sh all
 
-.PHONY: rabbit-clean
-rabbit-clean:
-	rabbitmqadmin -f pretty_json list queues vhost name messages | jq -r '.[] | "rabbitmqadmin delete queue name=\(.name) --vhost=\(.vhost)"' | bash
+.PHONY: clean-rabbit
+clean-rabbit:
+	rabbitmqadmin -f pretty_json list queues vhost name \
+		| jq -r '.[] | "rabbitmqadmin delete queue name=\(.name) --vhost=\(.vhost)"' \
+		| bash
+	rabbitmqadmin -f pretty_json list exchanges name vhost \
+		| jq -r '.[] |select(.name | startswith("amq") | not) | select (.name != "") | "rabbitmqadmin delete exchange name=\(.name) --vhost=\(.vhost)"' \
+		| bash
 
 # Clean
 .PHONY: full-clean
@@ -85,7 +90,7 @@ cabal.project.local:
 c: treefmt c-fast
 
 .PHONY: c
-c-fast: 
+c-fast:
 	cabal build $(WIRE_CABAL_BUILD_OPTIONS) $(package) || ( make clean-hint; false )
 ifeq ($(test), 1)
 	./hack/bin/cabal-run-tests.sh $(package) $(testargs)

changelog.d/0-release-notes/WBP-10308

@@ -0,0 +1 @@
+Notifications are now also sent via RabbitMQ therefore RabbitMQ is now a required configuration in brig.

deploy/dockerephemeral/docker-compose.yaml

@@ -271,6 +271,7 @@ services:
     ports:
       - '127.0.0.1:5671:5671'
       - '127.0.0.1:15671:15671'
+      - '127.0.0.1:15672:15672'
     volumes:
       - ./rabbitmq-config/rabbitmq.conf:/etc/rabbitmq/conf.d/20-wire.conf
       - ./rabbitmq-config/certificates:/etc/rabbitmq/certificates

integration/integration.cabal

@@ -122,6 +122,7 @@ library
     Test.Demo
     Test.EJPD
     Test.Errors
+    Test.Events
     Test.ExternalPartner
     Test.FeatureFlags
     Test.FeatureFlags.AppLock

integration/test/API/Galley.hs

@@ -733,3 +733,12 @@ getTeamMembersCsv :: (HasCallStack, MakesValue user) => user -> String -> App Re
 getTeamMembersCsv user tid = do
   req <- baseRequest user Galley Versioned (joinHttpPath ["teams", tid, "members", "csv"])
   submit "GET" req
+
+-- | https://staging-nginz-https.zinfra.io/v6/api/swagger-ui/#/default/post_conversations__cnv_domain___cnv__typing
+sendTypingStatus :: (HasCallStack, MakesValue user, MakesValue conv) => user -> conv -> String -> App Response
+sendTypingStatus user conv status = do
+  convDomain <- objDomain conv
+  convId <- objId conv
+  req <- baseRequest user Galley Versioned (joinHttpPath ["conversations", convDomain, convId, "typing"])
+  submit "POST"
+    $ addJSONObject ["status" .= status] req

integration/test/Notifications.hs

@@ -47,7 +47,7 @@ assertNoNotifications u uc since0 p = do
 awaitNotifications ::
   (HasCallStack, MakesValue user, MakesValue client) =>
   user ->
-  client ->
+  Maybe client ->
   Maybe String ->
   -- | Max no. of notifications
   Int ->
@@ -62,11 +62,11 @@ awaitNotifications user client since0 n selector = do
       | timeRemaining <= 0 = pure res0
       | otherwise =
           do
-            c <- make client & asString
+            c <- for client (asString . make)
             notifs <-
               getNotifications
                 user
-                def {since = since, client = Just c}
+                def {since = since, client = c}
                 `bindResponse` \resp -> asList (resp.json %. "notifications")
             lastNotifId <- case notifs of
               [] -> pure since
@@ -85,16 +85,26 @@ awaitNotifications user client since0 n selector = do
                 threadDelay 1_000
                 go (timeRemaining - 1) lastNotifId res
 
-awaitNotification ::
+awaitNotificationClient ::
   (HasCallStack, MakesValue user, MakesValue client, MakesValue lastNotifId) =>
   user ->
   client ->
   Maybe lastNotifId ->
   (Value -> App Bool) ->
   App Value
-awaitNotification user client lastNotifId selector = do
+awaitNotificationClient user client lastNotifId selector = do
+  since0 <- mapM objId lastNotifId
+  head <$> awaitNotifications user (Just client) since0 1 selector
+
+awaitNotification ::
+  (HasCallStack, MakesValue user, MakesValue lastNotifId) =>
+  user ->
+  Maybe lastNotifId ->
+  (Value -> App Bool) ->
+  App Value
+awaitNotification user lastNotifId selector = do
   since0 <- mapM objId lastNotifId
-  head <$> awaitNotifications user client since0 1 selector
+  head <$> awaitNotifications user (Nothing :: Maybe ()) since0 1 selector
 
 isDeleteUserNotif :: (MakesValue a) => a -> App Bool
 isDeleteUserNotif n =
@@ -219,9 +229,9 @@ assertLeaveNotification ::
   App ()
 assertLeaveNotification fromUser conv user client leaver =
   void
-    $ awaitNotification
+    $ awaitNotificationClient
       user
-      client
+      (Just client)
       noValue
       ( allPreds
           [ isConvLeaveNotif,

integration/test/SetupHelpers.hs

@@ -403,3 +403,13 @@ uploadDownloadProfilePicture :: (HasCallStack, MakesValue usr) => usr -> App (St
 uploadDownloadProfilePicture usr = do
   (dom, key, _payload) <- uploadProfilePicture usr
   downloadProfilePicture usr dom key
+
+mkContextUserIds :: (MakesValue user) => [(String, user)] -> App String
+mkContextUserIds =
+  fmap (intercalate "\n")
+    . traverse
+      ( \(name, user) -> do
+          uid <- objQidObject user %. "id" & asString
+          domain <- objDomain user
+          pure $ name <> ": " <> uid <> "@" <> domain
+      )

integration/test/Test/Conversation.hs

@@ -492,26 +492,25 @@ testSynchroniseUserRemovalNotification domain = do
   otherDomain <- make domain
   [alice, bob] <- createAndConnectUsers [ownDomain, otherDomain]
   runCodensity (acquireResources 1 resourcePool) $ \[dynBackend] -> do
-    (conv, charlie, client) <-
+    (conv, charlie) <-
       runCodensity (startDynamicBackend dynBackend mempty) $ \_ -> do
         charlie <- randomUser dynBackend.berDomain def
-        client <- objId $ bindResponse (addClient charlie def) $ getJSON 201
         mapM_ (connectTwoUsers charlie) [alice, bob]
         conv <-
           postConversation alice (defProteus {qualifiedUsers = [bob, charlie]})
             >>= getJSON 201
-        pure (conv, charlie, client)
+        pure (conv, charlie)
 
     let newConvName = "The new conversation name"
     bindResponse (changeConversationName alice conv newConvName) $ \resp ->
       resp.status `shouldMatchInt` 200
     bindResponse (removeMember alice conv charlie) $ \resp ->
       resp.status `shouldMatchInt` 200
     runCodensity (startDynamicBackend dynBackend mempty) $ \_ -> do
-      nameNotif <- awaitNotification charlie client noValue isConvNameChangeNotif
+      nameNotif <- awaitNotification charlie noValue isConvNameChangeNotif
       nameNotif %. "payload.0.qualified_conversation" `shouldMatch` objQidObject conv
       nameNotif %. "payload.0.data.name" `shouldMatch` newConvName
-      leaveNotif <- awaitNotification charlie client noValue isConvLeaveNotif
+      leaveNotif <- awaitNotification charlie noValue isConvLeaveNotif
       leaveNotif %. "payload.0.qualified_conversation" `shouldMatch` objQidObject conv
 
 testConvRenaming :: (HasCallStack) => App ()
@@ -648,19 +647,18 @@ testDeleteTeamConversationWithUnreachableRemoteMembers = do
         notif %. "payload.0.qualified_from" `shouldMatch` objQidObject alice
 
   runCodensity (acquireResources 1 resourcePool) $ \[dynBackend] -> do
-    (bob, bobClient) <- runCodensity (startDynamicBackend dynBackend mempty) $ \_ -> do
+    bob <- runCodensity (startDynamicBackend dynBackend mempty) $ \_ -> do
       bob <- randomUser dynBackend.berDomain def
-      bobClient <- objId $ bindResponse (addClient bob def) $ getJSON 201
       connectTwoUsers alice bob
       mem <- bob %. "qualified_id"
       void $ addMembers alice conv def {users = [mem]} >>= getBody 200
-      pure (bob, bobClient)
+      pure bob
     withWebSocket alice $ \ws -> do
       void $ deleteTeamConversation team conv alice >>= getBody 200
       notif <- awaitMatch isConvDeleteNotif ws
       assertNotification notif
     void $ runCodensity (startDynamicBackend dynBackend mempty) $ \_ -> do
-      notif <- awaitNotification bob bobClient noValue isConvDeleteNotif
+      notif <- awaitNotification bob noValue isConvDeleteNotif
       assertNotification notif
 
 testDeleteTeamMemberLimitedEventFanout :: (HasCallStack) => App ()