Add test for the temp queue / create the temp queue

wireapp · Dec 11, 2024 · a0fefcf · a0fefcf
1 parent fb85f1b
commit a0fefcf
Show file tree

Hide file tree

Showing 2 changed files with 51 additions and 21 deletions.
diff --git a/integration/test/Test/Events.hs b/integration/test/Test/Events.hs
@@ -32,7 +32,7 @@ testConsumeEventsOneWebSocket = do
   client <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
   clientId <- objId client
 
-  runCodensity (createEventsWebSocket alice clientId) $ \ws -> do
+  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
     deliveryTag <- assertEvent ws $ \e -> do
       e %. "type" `shouldMatch` "event"
       e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
@@ -57,6 +57,34 @@ testConsumeEventsOneWebSocket = do
     resp.status `shouldMatchInt` 200
     shouldBeEmpty $ resp.json %. "notifications"
 
+testConsumeTempEvents :: (HasCallStack) => App ()
+testConsumeTempEvents = do
+  alice <- randomUser OwnDomain def
+
+  client0 <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
+  runCodensity (createEventsWebSocket alice Nothing) $ \ws -> do
+    clientId <- objId client0
+
+    void $ assertEvent ws $ \e -> do
+      e %. "type" `shouldMatch` "event"
+      e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
+      e %. "data.event.payload.0.client.id" `shouldMatch` clientId
+
+      ackEvent ws e
+
+  runCodensity (createEventsWebSocket alice Nothing) $ \ws -> do
+    client <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
+    clientId <- objId client
+
+    void $ assertEvent ws $ \e -> do
+      e %. "type" `shouldMatch` "event"
+      e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
+      e %. "data.event.payload.0.client.id" `shouldMatch` clientId
+
+      ackEvent ws e
+
+    assertNoEvent ws
+
 testConsumeEventsForDifferentUsers :: (HasCallStack) => App ()
 testConsumeEventsForDifferentUsers = do
   alice <- randomUser OwnDomain def
@@ -69,8 +97,8 @@ testConsumeEventsForDifferentUsers = do
   bobClientId <- objId bobClient
 
   lowerCodensity $ do
-    aliceWS <- createEventsWebSocket alice aliceClientId
-    bobWS <- createEventsWebSocket bob bobClientId
+    aliceWS <- createEventsWebSocket alice (Just aliceClientId)
+    bobWS <- createEventsWebSocket bob (Just bobClientId)
     lift $ assertClientAdd aliceClientId aliceWS
     lift $ assertClientAdd bobClientId bobWS
   where
@@ -106,7 +134,7 @@ testConsumeEventsWhileHavingLegacyClients = do
     oldNotif <- awaitMatch isUserClientAddNotif oldWS
     oldNotif %. "payload.0.client.id" `shouldMatch` newClientId
 
-    runCodensity (createEventsWebSocket alice newClientId) $ \ws ->
+    runCodensity (createEventsWebSocket alice (Just newClientId)) $ \ws ->
       assertEvent ws $ \e -> do
         e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
         e %. "data.event.payload.0.client.id" `shouldMatch` newClientId
@@ -123,20 +151,20 @@ testConsumeEventsAcks = do
   client <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
   clientId <- objId client
 
-  runCodensity (createEventsWebSocket alice clientId) $ \ws -> do
+  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
     assertEvent ws $ \e -> do
       e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
       e %. "data.event.payload.0.client.id" `shouldMatch` clientId
 
   -- without ack, we receive the same event again
-  runCodensity (createEventsWebSocket alice clientId) $ \ws -> do
+  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
     deliveryTag <- assertEvent ws $ \e -> do
       e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
       e %. "data.event.payload.0.client.id" `shouldMatch` clientId
       e %. "data.delivery_tag"
     sendAck ws deliveryTag False
 
-  runCodensity (createEventsWebSocket alice clientId) $ \ws -> do
+  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
     assertNoEvent ws
 
 testConsumeEventsMultipleAcks :: (HasCallStack) => App ()
@@ -148,7 +176,7 @@ testConsumeEventsMultipleAcks = do
   handle <- randomHandle
   putHandle alice handle >>= assertSuccess
 
-  runCodensity (createEventsWebSocket alice clientId) $ \ws -> do
+  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
     assertEvent ws $ \e -> do
       e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
       e %. "data.event.payload.0.client.id" `shouldMatch` clientId
@@ -160,7 +188,7 @@ testConsumeEventsMultipleAcks = do
 
     sendAck ws deliveryTag True
 
-  runCodensity (createEventsWebSocket alice clientId) $ \ws -> do
+  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
     assertNoEvent ws
 
 testConsumeEventsAckNewEventWithoutAckingOldOne :: (HasCallStack) => App ()
@@ -172,7 +200,7 @@ testConsumeEventsAckNewEventWithoutAckingOldOne = do
   handle <- randomHandle
   putHandle alice handle >>= assertSuccess
 
-  runCodensity (createEventsWebSocket alice clientId) $ \ws -> do
+  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
     assertEvent ws $ \e -> do
       e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
       e %. "data.event.payload.0.client.id" `shouldMatch` clientId
@@ -186,15 +214,15 @@ testConsumeEventsAckNewEventWithoutAckingOldOne = do
     sendAck ws deliveryTagHandleAdd False
 
   -- Expect client-add event to be delivered again.
-  runCodensity (createEventsWebSocket alice clientId) $ \ws -> do
+  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
     deliveryTagClientAdd <- assertEvent ws $ \e -> do
       e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
       e %. "data.event.payload.0.client.id" `shouldMatch` clientId
       e %. "data.delivery_tag"
 
     sendAck ws deliveryTagClientAdd False
 
-  runCodensity (createEventsWebSocket alice clientId) $ \ws -> do
+  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
     assertNoEvent ws
 
 testEventsDeadLettered :: (HasCallStack) => App ()
@@ -214,7 +242,7 @@ testEventsDeadLettered = do
     handle1 <- randomHandle
     putHandle alice handle1 >>= assertSuccess
 
-    runCodensity (createEventsWebSocket alice clientId) $ \ws -> do
+    runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
       assertEvent ws $ \e -> do
         e %. "type" `shouldMatch` "notifications.missed"
 
@@ -241,7 +269,7 @@ testTransientEventsDoNotTriggerDeadLetters = do
     clientId <- objId client
 
     -- consume it
-    runCodensity (createEventsWebSocket alice clientId) $ \ws -> do
+    runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
       assertEvent ws $ \e -> do
         e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
         e %. "type" `shouldMatch` "event"
@@ -256,7 +284,7 @@ testTransientEventsDoNotTriggerDeadLetters = do
     -- Typing status is transient, currently no one is listening.
     sendTypingStatus alice selfConvId "started" >>= assertSuccess
 
-    runCodensity (createEventsWebSocket alice clientId) $ \ws -> do
+    runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
       assertNoEvent ws
 
 testTransientEvents :: (HasCallStack) => App ()
@@ -269,7 +297,7 @@ testTransientEvents = do
   -- indicators, so we don't have to create another conv.
   selfConvId <- objQidObject alice
 
-  runCodensity (createEventsWebSocket alice clientId) $ \ws -> do
+  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
     consumeAllEvents ws
     sendTypingStatus alice selfConvId "started" >>= assertSuccess
     assertEvent ws $ \e -> do
@@ -289,7 +317,7 @@ testTransientEvents = do
   -- We shouldn't see the stopped typing status because we were not connected to
   -- the websocket when it was sent. The other events should still show up in
   -- order.
-  runCodensity (createEventsWebSocket alice clientId) $ \ws -> do
+  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
     for_ [handle1, handle2] $ \handle ->
       assertEvent ws $ \e -> do
         e %. "data.event.payload.0.type" `shouldMatch` "user.update"
@@ -317,7 +345,7 @@ testChannelLimit = withModifiedBackend
 
     lowerCodensity $ do
       for_ clients $ \c -> do
-        ws <- createEventsWebSocket alice c
+        ws <- createEventsWebSocket alice (Just c)
         e <- Codensity $ \k -> assertEvent ws k
         lift $ do
           e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
@@ -326,7 +354,7 @@ testChannelLimit = withModifiedBackend
 
       -- the first client fails to connect because the server runs out of channels
       do
-        ws <- createEventsWebSocket alice client0
+        ws <- createEventsWebSocket alice (Just client0)
         lift $ assertNoEvent ws
 
 ----------------------------------------------------------------------
@@ -340,15 +368,15 @@ data EventWebSocket = EventWebSocket
 createEventsWebSocket ::
   (HasCallStack, MakesValue uid) =>
   uid ->
-  String ->
+  Maybe String ->
   Codensity App EventWebSocket
 createEventsWebSocket user cid = do
   eventsChan <- liftIO newChan
   ackChan <- liftIO newEmptyMVar
   serviceMap <- lift $ getServiceMap =<< objDomain user
   uid <- lift $ objId =<< objQidObject user
   let HostPort caHost caPort = serviceHostPort serviceMap Cannon
-      path = "/events?client=" <> cid
+      path = "/events" <> maybe "" ("?client=" <>) cid
       caHdrs = [(fromString "Z-User", toByteString' uid)]
       app conn =
         race_

diff --git a/services/cannon/src/Cannon/RabbitMqConsumerApp.hs b/services/cannon/src/Cannon/RabbitMqConsumerApp.hs
@@ -15,6 +15,7 @@ import Data.Aeson hiding (Key)
 import Data.Id
 import Data.Text qualified as T
 import Imports hiding (min, threadDelay)
+import Network.AMQP (newQueue)
 import Network.AMQP qualified as Q
 import Network.WebSockets
 import Network.WebSockets qualified as WS
@@ -117,6 +118,7 @@ rabbitMQWebSocketApp uid mcid e pendingConn = do
       let createQueue chan = case mcid of
             Nothing -> Codensity $ \k ->
               ( do
+                  void $ Q.declareQueue chan newQueue {Q.queueName = queueName}
                   for_ [userRoutingKey uid, temporaryRoutingKey uid] $
                     Q.bindQueue chan queueName userNotificationExchangeName
                   k ()