Do not deliver client specific notifiations to temporary clients (#4703)

* gundeck: Never push a notification directly to temp clients

Also: Only push a notification to RabbitMQ when the user has any clients which
support consumable notifications.

This way RabbitMQ wouldn't think there are notifications meant for no queues.

* Delete unnecessary constructor RecipientClientsTemporaryOnly

* changelog

* Apply suggestions from code review

Co-authored-by: Sven Tennie <[email protected]>

---------

Co-authored-by: Sven Tennie <[email protected]>

Author: akshaymankar

changelog.d/2-features/fewer-events-for-temp-clients

@@ -0,0 +1 @@
+Do not deliver client specific notifications to temporary clients.

integration/test/Test/Events.hs

@@ -21,7 +21,6 @@ import Data.ProtoLens.Labels ()
 import Data.Proxy (Proxy (..))
 import qualified Data.Text as Text
 import Data.Timeout
-import MLS.Util
 import Network.AMQP.Extended
 import Network.RabbitMqAdmin
 import qualified Network.WebSockets as WS
@@ -142,43 +141,35 @@ testConsumeTempEvents = do
 
         ackEvent ws e
 
-testConsumeTempEventsWithoutOwnClient :: (HasCallStack) => App ()
-testConsumeTempEventsWithoutOwnClient = do
-  [alice, bob] <- createAndConnectUsers [OwnDomain, OwnDomain]
-
-  runCodensity (createEventsWebSocket alice Nothing) $ \ws -> do
-    handle <- randomHandle
-    putHandle bob handle >>= assertSuccess
-
-    -- We cannot use 'assertEvent' here because there is a race between the temp
-    -- queue being created and rabbitmq fanning out the previous events.
-    void $ assertFindsEvent ws $ \e -> do
-      e %. "type" `shouldMatch` "event"
-      e %. "data.event.payload.0.type" `shouldMatch` "user.update"
-      e %. "data.event.payload.0.user.id" `shouldMatch` objId bob
-      e %. "data.event.payload.0.user.handle" `shouldMatch` handle
-
-      ackEvent ws e
-
 testTemporaryQueuesAreDeletedAfterUse :: (HasCallStack) => App ()
 testTemporaryQueuesAreDeletedAfterUse = do
   startDynamicBackendsReturnResources [def] $ \[beResource] -> do
     let domain = beResource.berDomain
     rabbitmqAdmin <- mkRabbitMqAdminClientForResource beResource
-    queuesBeforeWS <- rabbitmqAdmin.listQueuesByVHost (fromString beResource.berVHost) (fromString "") True 100 1
-    let deadNotifsQueue = Queue {name = fromString "dead-user-notifications", vhost = fromString beResource.berVHost}
+    [alice, bob] <- createAndConnectUsers [domain, domain]
+
+    -- Create client for alice, so their temp websocket works
+    aliceClient <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
+    aliceId <- asString $ alice %. "qualified_id.id"
+    aliceClientId <- asString $ aliceClient %. "id"
+
+    let aliceClientQueueName = "user-notifications." <> aliceId <> "." <> aliceClientId
+        aliceClientQueue = Queue {name = fromString aliceClientQueueName, vhost = fromString beResource.berVHost}
+        deadNotifsQueue = Queue {name = fromString "dead-user-notifications", vhost = fromString beResource.berVHost}
         cellsEventsQueue = Queue {name = fromString "cells_events", vhost = fromString beResource.berVHost}
-    queuesBeforeWS.items `shouldMatchSet` [deadNotifsQueue, cellsEventsQueue]
 
-    [alice, bob] <- createAndConnectUsers [domain, domain]
+    -- Wait for queue for the new client to be created
+    eventually $ do
+      queuesBeforeWS <- rabbitmqAdmin.listQueuesByVHost (fromString beResource.berVHost) (fromString "") True 100 1
+      queuesBeforeWS.items `shouldMatchSet` [deadNotifsQueue, cellsEventsQueue, aliceClientQueue]
 
     runCodensity (createEventsWebSocket alice Nothing) $ \ws -> do
       handle <- randomHandle
       putHandle bob handle >>= assertSuccess
 
       queuesDuringWS <- rabbitmqAdmin.listQueuesByVHost (fromString beResource.berVHost) (fromString "") True 100 1
       addJSONToFailureContext "queuesDuringWS" queuesDuringWS $ do
-        length queuesDuringWS.items `shouldMatchInt` 3
+        length queuesDuringWS.items `shouldMatchInt` 4
 
       -- We cannot use 'assertEvent' here because there is a race between the temp
       -- queue being created and rabbitmq fanning out the previous events.
@@ -190,49 +181,9 @@ testTemporaryQueuesAreDeletedAfterUse = do
 
         ackEvent ws e
 
-    -- Use let binding here so 'shouldMatchEventually' retries the whole request
-    let queuesAfterWSM = rabbitmqAdmin.listQueuesByVHost (fromString beResource.berVHost) (fromString "") True 100 1
-    eventually $ fmap (.items) queuesAfterWSM `shouldMatchSet` [deadNotifsQueue, cellsEventsQueue]
-
-testMLSTempEvents :: (HasCallStack) => App ()
-testMLSTempEvents = do
-  [alice, bob] <- createAndConnectUsers [OwnDomain, OwnDomain]
-  clients@[alice1, _, _] <-
-    traverse
-      ( createMLSClient
-          def
-            { clientArgs =
-                def
-                  { acapabilities = Just ["consumable-notifications"]
-                  }
-            }
-      )
-      [alice, bob, bob]
-
-  traverse_ (uploadNewKeyPackage def) clients
-  convId <- createNewGroup def alice1
-
-  runCodensity (createEventsWebSocket bob Nothing) $ \ws -> do
-    commit <- createAddCommit alice1 convId [bob]
-    void $ postMLSCommitBundle commit.sender (mkBundle commit) >>= getJSON 201
-
-    -- FUTUREWORK: we should not rely on events arriving in this particular order
-
-    -- We cannot use 'assertEvent' here because there is a race between the temp
-    -- queue being created and rabbitmq fanning out the previous events.
-    void $ assertFindsEvent ws $ \e -> do
-      e %. "type" `shouldMatch` "event"
-      e %. "data.event.payload.0.type" `shouldMatch` "conversation.member-join"
-      user <- assertOne =<< (e %. "data.event.payload.0.data.users" & asList)
-      user %. "qualified_id" `shouldMatch` (bob %. "qualified_id")
-      ackEvent ws e
-
-    void $ assertEvent ws $ \e -> do
-      e %. "type" `shouldMatch` "event"
-      e %. "data.event.payload.0.type" `shouldMatch` "conversation.mls-welcome"
-      ackEvent ws e
-
-    assertNoEvent_ ws
+    eventually $ do
+      queuesAfterWS <- rabbitmqAdmin.listQueuesByVHost (fromString beResource.berVHost) (fromString "") True 100 1
+      queuesAfterWS.items `shouldMatchSet` [deadNotifsQueue, cellsEventsQueue, aliceClientQueue]
 
 testSendMessageNoReturnToSenderWithConsumableNotificationsProteus :: (HasCallStack) => App ()
 testSendMessageNoReturnToSenderWithConsumableNotificationsProteus = do
@@ -280,7 +231,7 @@ testEventsForSpecificClients = do
     ws1 <- createEventsWebSocket alice (Just cid1)
     wsTemp <- createEventsWebSocket alice Nothing
     lift $ do
-      forM_ [ws1, wsTemp] consumeAllEvents
+      void $ consumeAllEvents ws1
 
       let eventForClient1 =
             object
@@ -297,17 +248,12 @@ testEventsForSpecificClients = do
       assertEvent ws1 $ \e ->
         e %. "data.event.payload.0.hello" `shouldMatch` "client1"
 
-      assertEvent wsTemp $ \e -> do
-        e %. "data.event.payload.0.hello" `shouldMatch` "client1"
-        ackEvent wsTemp e
-
-      assertEvent wsTemp $ \e -> do
-        e %. "data.event.payload.0.hello" `shouldMatch` "client2"
-        ackEvent wsTemp e
-
       addFailureContext "client 1 should not get any events meant for client 2"
         $ assertNoEvent_ ws1
 
+      addFailureContext "temp client should not get any events meant solely for client 1 or 2"
+        $ assertNoEvent_ wsTemp
+
 testConsumeEventsForDifferentUsers :: (HasCallStack) => App ()
 testConsumeEventsForDifferentUsers = do
   alice <- randomUser OwnDomain def

libs/wire-api/src/Wire/API/Push/V2.hs

@@ -112,20 +112,9 @@ data RecipientClients
     RecipientClientsAll
   | -- | An explicit list of clients
     RecipientClientsSome (List1 ClientId)
-  | -- | Only temporary clients receive these events.
-    -- It not supposed to be used by consumers of gundeck, only here for
-    -- internal gundeck use.
-    --
-    -- This is a little hack to make the 'splitPush' function work properly so
-    -- that temporary clients receive notifications over rabbitmq even when there
-    -- are no real clients which support consumable-notifications. This should
-    -- go away when we drop support for Cassandra.
-    RecipientClientsTemporaryOnly
   deriving (Eq, Show, Ord, Generic)
   deriving (FromJSON, ToJSON, S.ToSchema) via (Schema RecipientClients)
 
--- | This doesn't produce the 'RecipientClientsTemporaryOnly' case becasue we
--- don't expect it to come from outside gundeck.
 instance Arbitrary RecipientClients where
   arbitrary =
     oneof [allClients, someClients]
@@ -136,15 +125,6 @@ instance Arbitrary RecipientClients where
           firstClientId <- arbitrary
           (List1.list1 firstClientId . filter (/= firstClientId) . Set.toList <$> setOf' arbitrary)
 
-instance Semigroup RecipientClients where
-  RecipientClientsAll <> _ = RecipientClientsAll
-  _ <> RecipientClientsAll = RecipientClientsAll
-  RecipientClientsSome cs1 <> RecipientClientsSome cs2 =
-    RecipientClientsSome (cs1 <> cs2)
-  RecipientClientsTemporaryOnly <> x =
-    x
-  x <> RecipientClientsTemporaryOnly = x
-
 instance ToSchema Recipient where
   schema =
     object "Recipient" $
@@ -163,7 +143,6 @@ instance ToSchema RecipientClients where
           & (S.schema . S.description ?~ "List of clientIds. Empty means `all clients`.")
 
       i :: A.Value -> A.Parser RecipientClients
-      i (A.String "temp-only") = pure RecipientClientsTemporaryOnly
       i v =
         parseJSON @[ClientId] v >>= \case
           [] -> pure RecipientClientsAll
@@ -174,7 +153,6 @@ instance ToSchema RecipientClients where
         pure . \case
           RecipientClientsSome cs -> toJSON cs
           RecipientClientsAll -> A.Array mempty
-          RecipientClientsTemporaryOnly -> A.String "temp-only"
 
 makeLenses ''Recipient
 

services/gundeck/src/Gundeck/Push.hs

@@ -74,7 +74,6 @@ import Util.Options
 import Wire.API.Internal.Notification
 import Wire.API.Notification
 import Wire.API.Presence (Presence (..))
-import Wire.API.Presence qualified as Presence
 import Wire.API.Push.Token qualified as Public
 import Wire.API.Push.V2
 import Wire.API.User (UserSet (..))
@@ -146,17 +145,19 @@ instance MonadMapAsync Gundeck where
       Nothing -> mapAsync f l
       Just chunkSize -> concat <$> mapM (mapAsync f) (List.chunksOf chunkSize l)
 
-splitPushes :: (MonadPushAll m) => [Push] -> m ([Push], [Push])
+splitPushes :: (MonadPushAll m) => [Push] -> m ([Push], [Push], UserClientsFull)
 splitPushes ps = do
   allUserClients <- mpaGetClients (Set.unions $ map (\p -> Set.map (._recipientId) $ p._pushRecipients) ps)
-  pure . partitionHereThere $ map (splitPush allUserClients) ps
+  let (rabbitmqPushes, legacyPushes) = partitionHereThere $ map (splitPush allUserClients) ps
+  pure (rabbitmqPushes, legacyPushes, allUserClients)
 
 -- | Split a push into rabbitmq and legacy push. This code exists to help with
 -- migration. Once it is completed and old APIs are not supported anymore we can
 -- assume everything is meant for RabbitMQ and stop splitting.
 splitPush ::
   UserClientsFull ->
   Push ->
+  -- | These rabbitmqPush cassandraPush
   These Push Push
 splitPush clientsFull p = do
   let (rabbitmqRecipients, legacyRecipients) =
@@ -177,7 +178,6 @@ splitPush clientsFull p = do
             RecipientClientsSome cs ->
               Set.filter (\c -> c.clientId `elem` toList cs) allClients
             RecipientClientsAll -> allClients
-            RecipientClientsTemporaryOnly -> mempty
           (rabbitmqClients, legacyClients) = Set.partition supportsConsumableNotifications relevantClients
           rabbitmqClientIds = (.clientId) <$> Set.toList rabbitmqClients
           legacyClientIds = (.clientId) <$> Set.toList legacyClients
@@ -190,13 +190,16 @@ splitPush clientsFull p = do
           -- We return all clients for RabbitMQ even if there are no real
           -- clients so a temporary client can still read the notifications on
           -- RabbitMQ.
-          (These rcpt {_recipientClients = RecipientClientsTemporaryOnly} rcpt)
+          That rcpt
         (_, []) ->
-          (This rcpt)
+          This rcpt
         (r : rs, l : ls) ->
-          These
-            rcpt {_recipientClients = RecipientClientsSome $ list1 r rs}
-            rcpt {_recipientClients = RecipientClientsSome $ list1 l ls}
+          let rabbitMqRecipients = case rcpt._recipientClients of
+                RecipientClientsAll -> RecipientClientsAll
+                RecipientClientsSome _ -> RecipientClientsSome $ list1 r rs
+           in These
+                rcpt {_recipientClients = rabbitMqRecipients}
+                rcpt {_recipientClients = RecipientClientsSome $ list1 l ls}
 
 getClients :: Set UserId -> Gundeck UserClientsFull
 getClients uids = do
@@ -221,13 +224,13 @@ getClients uids = do
 pushAll :: (MonadPushAll m, MonadNativeTargets m, MonadMapAsync m, Log.MonadLogger m) => [Push] -> m ()
 pushAll pushes = do
   Log.debug $ msg (val "pushing") . Log.field "pushes" (Aeson.encode pushes)
-  (rabbitmqPushes, legacyPushes) <- splitPushes pushes
+  (rabbitmqPushes, legacyPushes, allUserClients) <- splitPushes pushes
 
   legacyNotifs <- mapM mkNewNotification legacyPushes
-  pushAllLegacy legacyNotifs
+  pushAllLegacy legacyNotifs allUserClients
 
   rabbitmqNotifs <- mapM mkNewNotification rabbitmqPushes
-  pushAllViaRabbitMq rabbitmqNotifs
+  pushAllViaRabbitMq rabbitmqNotifs allUserClients
 
   -- Note that Cells needs all notifications because it doesn't matter whether
   -- some recipients have rabbitmq clients or not.
@@ -240,8 +243,8 @@ pushAll pushes = do
 
 -- | Construct and send a single bulk push request to the client.  Write the 'Notification's from
 -- the request to C*.  Trigger native pushes for all delivery failures notifications.
-pushAllLegacy :: (MonadPushAll m, MonadNativeTargets m, MonadMapAsync m) => [NewNotification] -> m ()
-pushAllLegacy newNotifications = do
+pushAllLegacy :: (MonadPushAll m, MonadNativeTargets m, MonadMapAsync m) => [NewNotification] -> UserClientsFull -> m ()
+pushAllLegacy newNotifications userClientsFull = do
   -- persist push request
   let cassandraTargets :: [CassandraTargets]
       cassandraTargets = map mkCassandraTargets newNotifications
@@ -257,11 +260,14 @@ pushAllLegacy newNotifications = do
     wsTargets <- mapM mkWSTargets newNotifications
     resp <- compilePushResps wsTargets <$> mpaBulkPush (compilePushReq <$> wsTargets)
     -- native push
-    forM_ resp $ \((notif :: Notification, psh :: Push), alreadySent :: [Presence]) ->
-      pushNativeWithBudget notif psh alreadySent
-
-pushNativeWithBudget :: (MonadMapAsync m, MonadPushAll m, MonadNativeTargets m) => Notification -> Push -> [Presence] -> m ()
-pushNativeWithBudget notif psh alreadySent = do
+    forM_ resp $ \((notif :: Notification, psh :: Push), alreadySent :: [Presence]) -> do
+      let alreadySentClients = Set.fromList $ mapMaybe (\p -> (p.userId,) <$> p.clientId) alreadySent
+          rabbitmqClients = Map.map (Set.filter supportsConsumableNotifications) userClientsFull.userClientsFull
+          rabbitmqClientIds = Map.foldMapWithKey (\uid clients -> Set.map (\c -> (uid, c.clientId)) clients) rabbitmqClients
+      pushNativeWithBudget notif psh (Set.toList $ Set.union alreadySentClients rabbitmqClientIds)
+
+pushNativeWithBudget :: (MonadMapAsync m, MonadPushAll m, MonadNativeTargets m) => Notification -> Push -> [(UserId, ClientId)] -> m ()
+pushNativeWithBudget notif psh dontPush = do
   perPushConcurrency <- mntgtPerPushConcurrency
   let rcps' = nativeTargetsRecipients psh
       cost = maybe (length rcps') (min (length rcps')) perPushConcurrency
@@ -272,14 +278,16 @@ pushNativeWithBudget notif psh alreadySent = do
   -- to cassandra and SNS are limited to 'perNativePushConcurrency' in parallel.
   unless (psh ^. pushTransient) $
     mpaRunWithBudget cost () $
-      mpaPushNative notif (psh ^. pushNativePriority) =<< nativeTargets psh rcps' alreadySent
+      mpaPushNative notif (psh ^. pushNativePriority) =<< nativeTargets psh rcps' dontPush
 
-pushAllViaRabbitMq :: (MonadPushAll m, MonadMapAsync m, MonadNativeTargets m) => [NewNotification] -> m ()
-pushAllViaRabbitMq newNotifs = do
+pushAllViaRabbitMq :: (MonadPushAll m, MonadMapAsync m, MonadNativeTargets m) => [NewNotification] -> UserClientsFull -> m ()
+pushAllViaRabbitMq newNotifs userClientsFull = do
   for_ newNotifs $ pushViaRabbitMq
   mpaForkIO $ do
-    for_ newNotifs $ \newNotif ->
-      pushNativeWithBudget newNotif.nnNotification newNotif.nnPush []
+    for_ newNotifs $ \newNotif -> do
+      let cassandraClients = Map.map (Set.filter $ not . supportsConsumableNotifications) userClientsFull.userClientsFull
+          cassandraClientIds = Map.foldMapWithKey (\uid clients -> Set.map (\c -> (uid, c.clientId)) clients) cassandraClients
+      pushNativeWithBudget newNotif.nnNotification newNotif.nnPush (Set.toList $ cassandraClientIds)
 
 pushViaRabbitMq :: (MonadPushAll m) => NewNotification -> m ()
 pushViaRabbitMq newNotif = do
@@ -291,11 +299,7 @@ pushViaRabbitMq newNotif = do
               RecipientClientsAll ->
                 Set.singleton $ userRoutingKey r._recipientId
               RecipientClientsSome (toList -> cs) ->
-                Set.fromList $
-                  temporaryRoutingKey r._recipientId
-                    : map (clientRoutingKey r._recipientId) cs
-              RecipientClientsTemporaryOnly ->
-                Set.singleton $ temporaryRoutingKey r._recipientId
+                Set.fromList $ map (clientRoutingKey r._recipientId) cs
   for_ routingKeys $ \routingKey ->
     mpaPublishToRabbitMq userNotificationExchangeName routingKey qMsg
 
@@ -382,7 +386,6 @@ mkCassandraTargets NewNotification {..} =
           -- clients are stored in cassandra as a list with a notification.  empty list
           -- is interpreted as "all clients" by 'Gundeck.Notification.Data.toNotif'.
           RecipientClientsSome cs -> Just $ toList cs
-          RecipientClientsTemporaryOnly -> Nothing
       pure $ target (r ^. recipientId) & targetClients .~ clients
 
 -- | Information needed to push notifications over websockets and/or native
@@ -465,9 +468,9 @@ nativeTargets ::
   (MonadNativeTargets m, MonadMapAsync m) =>
   Push ->
   [Recipient] ->
-  [Presence] ->
+  [(UserId, ClientId)] ->
   m [Address]
-nativeTargets psh rcps' alreadySent =
+nativeTargets psh rcps' dontPush =
   mntgtMapAsync addresses rcps' >>= fmap concat . mapM check
   where
     addresses :: Recipient -> m [Address]
@@ -483,14 +486,9 @@ nativeTargets psh rcps' alreadySent =
       -- Is the client not whitelisted?
       | not (whitelistedOrNoWhitelist a) = False
       -- Include client if not found in already served presences.
-      | otherwise = isNothing (List.find (isOnline a) alreadySent)
-    isOnline a x =
-      a ^. addrUser == Presence.userId x
-        && (a ^. addrConn == Presence.connId x || equalClient a x)
-    equalClient a x = Just (a ^. addrClient) == Presence.clientId x
+      | otherwise = not $ List.elem (a ^. addrUser, a ^. addrClient) dontPush --  (List.find (isOnline a) alreadySent)
     eligibleClient _ RecipientClientsAll = True
     eligibleClient a (RecipientClientsSome cs) = (a ^. addrClient) `elem` cs
-    eligibleClient _ RecipientClientsTemporaryOnly = False
     whitelistedOrNoWhitelist a =
       null (psh ^. pushConnections)
         || a ^. addrConn `elem` psh ^. pushConnections