WPB-28422 reconcile stale local memberships for deleted remote conversations

changelog.d/6-federation/WPB-28422

@@ -0,0 +1 @@
+Remove stale local memberships when a remote conversation is definitively reported as not found.

integration/test/Test/Conversation.hs

@@ -582,6 +582,189 @@ testGetOneOnOneConvInStatusSentFromRemote domain = do
   resp <- getConversation d1User d2ConvId
   resp.status `shouldMatchInt` 200
 
+testReconcileStaleLocalMembershipsForDeletedRemoteConversation :: (HasCallStack) => App ()
+testReconcileStaleLocalMembershipsForDeletedRemoteConversation = do
+  owner <- randomUser OwnDomain def
+  alice <- randomUser OtherDomain def
+  charlie <- randomUser OtherDomain def
+  for_ [alice, charlie] $ connectTwoUsers owner
+
+  conv <- registerMissingRemoteConversation owner [alice, charlie]
+
+  eventually $ do
+    assertConversationMembership alice conv True
+    assertConversationMembership charlie conv True
+
+  let isSystemDeleteFor conversation event =
+        fieldEquals event "payload.0.type" "conversation.system.delete"
+          &&~ isNotifConv conversation event
+
+  -- A successful response from the owning backend that omits the conversation
+  -- proves that Alice's locally stored membership is stale.
+  withWebSockets [alice, charlie] $ \[wsAlice, wsCharlie] -> do
+    getConversation alice conv >>= assertLabel 404 "no-conversation"
+    void $ awaitMatch (isSystemDeleteFor conv) wsAlice
+    assertConversationMembership alice conv False
+    assertConversationMembership charlie conv True
+
+    -- The bulk endpoint performs the same reconciliation for Charlie.
+    bindResponse (listConversations charlie [conv]) $ \resp -> do
+      resp.status `shouldMatchInt` 200
+      resp.json %. "found" `shouldMatch` ([] :: [Value])
+      resp.json %. "not_found" `shouldMatch` [conv]
+      resp.json %. "failed" `shouldMatch` ([] :: [Value])
+    void $ awaitMatch (isSystemDeleteFor conv) wsCharlie
+
+  assertConversationMembership alice conv False
+  assertConversationMembership charlie conv False
+
+  -- Reconciliation is idempotent once the local membership has been removed.
+  getConversation alice conv >>= assertLabel 404 "no-conversation"
+
+-- | Fetching stale remote conversations from two different domains reconciles
+-- both independently. A response for one domain must not remove memberships
+-- belonging to another domain.
+testReconcileStaleMembershipsMultipleDomains :: (HasCallStack) => App ()
+testReconcileStaleMembershipsMultipleDomains = do
+  resourcePool <- asks resourcePool
+  runCodensity (acquireResources 1 resourcePool) $ \[remoteBackend] ->
+    runCodensity (startDynamicBackend remoteBackend mempty) $ \_ -> do
+      alice <- randomUser OwnDomain def
+      ownerStatic <- randomUser OtherDomain def
+      ownerDynamic <- randomUser remoteBackend.berDomain def
+      connectTwoUsers ownerStatic alice
+      connectTwoUsers ownerDynamic alice
+      convStatic <- registerMissingRemoteConversation ownerStatic [alice]
+      convDynamic <- registerMissingRemoteConversation ownerDynamic [alice]
+
+      eventually $ do
+        assertConversationMembership alice convStatic True
+        assertConversationMembership alice convDynamic True
+
+      bindResponse (listConversations alice [convStatic, convDynamic]) $ \resp -> do
+        resp.status `shouldMatchInt` 200
+        resp.json %. "found" `shouldMatch` ([] :: [Value])
+        resp.json %. "failed" `shouldMatch` ([] :: [Value])
+        notFound <- resp.json %. "not_found" & asList
+        for_ [convStatic, convDynamic] $ \conv ->
+          (notFound :: [Value]) `shouldContain` [conv]
+
+      assertConversationMembership alice convStatic False
+      assertConversationMembership alice convDynamic False
+
+-- | Conversations the remote still returns are preserved; only omitted ones
+-- are reconciled within the same request.
+testReconcileOnlyMissingConversations :: (HasCallStack) => App ()
+testReconcileOnlyMissingConversations = do
+  alice <- randomUser OwnDomain def
+  owner <- randomUser OtherDomain def
+  connectTwoUsers owner alice
+
+  alive <-
+    postConversation owner (defProteus {qualifiedUsers = [alice]})
+      >>= getJSON 201
+  aliveQid <- objQidObject alive
+  stale <- registerMissingRemoteConversation owner [alice]
+
+  eventually $ do
+    assertConversationMembership alice aliveQid True
+    assertConversationMembership alice stale True
+
+  bindResponse (listConversations alice [aliveQid, stale]) $ \resp -> do
+    resp.status `shouldMatchInt` 200
+    resp.json %. "failed" `shouldMatch` ([] :: [Value])
+    found <- resp.json %. "found" & asList
+    length (found :: [Value]) `shouldMatchInt` 1
+    notFound <- resp.json %. "not_found" & asList
+    (notFound :: [Value]) `shouldContain` [stale]
+
+  assertConversationMembership alice aliveQid True
+  assertConversationMembership alice stale False
+
+testPreserveRemoteMembershipOnFederationFailure :: (HasCallStack) => App ()
+testPreserveRemoteMembershipOnFederationFailure = do
+  resourcePool <- asks resourcePool
+  runCodensity (acquireResources 1 resourcePool) $ \[remoteBackend] -> do
+    (alice, convQid) <- runCodensity (startDynamicBackend remoteBackend mempty) $ \_ -> do
+      owner <- randomUser remoteBackend.berDomain def
+      alice <- randomUser OwnDomain def
+      connectTwoUsers owner alice
+      conv <-
+        postConversation owner (defProteus {qualifiedUsers = [alice]})
+          >>= getJSON 201
+      convQid <- objQidObject conv
+      eventually $ assertConversationMembership alice convQid True
+      pure (alice, convQid)
+
+    bindResponse (listConversations alice [convQid]) $ \resp -> do
+      resp.status `shouldMatchInt` 200
+      resp.json %. "failed" `shouldMatch` [convQid]
+    assertConversationMembership alice convQid True
+
+-- | This function only submits the on-conversation-created event
+-- to the remote backend without actually having created a local conversation.
+registerMissingRemoteConversation :: (HasCallStack) => Value -> [Value] -> App Value
+registerMissingRemoteConversation owner members = do
+  originDomain <- objDomain owner
+  originUserId <- objId owner
+  convId <- randomId
+  targetDomain <- case members of
+    [] -> assertFailure "A remote conversation needs at least one local member"
+    firstMember : remainingMembers -> do
+      domain <- objDomain firstMember
+      for_ remainingMembers $ \remoteMember -> do
+        memberDomain <- objDomain remoteMember
+        memberDomain `shouldMatch` domain
+      pure domain
+  memberPayloads <- for members $ \remoteMember -> do
+    memberId <- objId remoteMember
+    memberQid <- objQidObject remoteMember
+    pure
+      $ object
+        [ "id" .= memberId,
+          "qualified_id" .= memberQid,
+          "status" .= (0 :: Int),
+          "conversation_role" .= ("wire_member" :: String)
+        ]
+  req <-
+    rawBaseRequest
+      originDomain
+      FederatorInternal
+      Unversioned
+      (joinHttpPath ["rpc", targetDomain, "galley", "on-conversation-created"])
+  bindResponse
+    ( submit "POST"
+        $ req
+        & addHeader "Wire-Origin-Domain" originDomain
+        & addJSONObject
+          [ "time" .= ("2026-01-01T00:00:00.000Z" :: String),
+            "orig_user_id" .= originUserId,
+            "cnv_id" .= convId,
+            "cnv_type" .= (0 :: Int),
+            "cnv_access" .= ["invite" :: String, "code"],
+            "cnv_access_roles" .= ["team_member" :: String, "non_team_member"],
+            "cnv_name" .= Aeson.Null,
+            "non_creator_members" .= memberPayloads,
+            "message_timer" .= Aeson.Null,
+            "receipt_mode" .= Aeson.Null,
+            "protocol" .= object ["protocol" .= ("proteus" :: String)],
+            "group_conv_type" .= ("group_conversation" :: String),
+            "channel_add_permission" .= Aeson.Null,
+            "history" .= Aeson.Null
+          ]
+    )
+    $ \resp -> resp.status `shouldMatchInt` 200
+  pure $ object ["domain" .= originDomain, "id" .= convId]
+
+assertConversationMembership :: (HasCallStack) => Value -> Value -> Bool -> App ()
+assertConversationMembership user conv expected =
+  bindResponse (listConversationIds user def) $ \resp -> do
+    resp.status `shouldMatchInt` 200
+    conversationIds <- resp.json %. "qualified_conversations" & asList
+    if expected
+      then conversationIds `shouldContain` [conv]
+      else conversationIds `shouldNotContain` [conv]
+
 testAddingUserNonFullyConnectedFederation :: (HasCallStack) => StaticDomain -> App ()
 testAddingUserNonFullyConnectedFederation domain = do
   let overrides =
@@ -1029,8 +1212,13 @@ testOnUserDeletedConversations = do
 
       do
         -- Bob is not in the one-to-one conversation with Alice any more
-        conv <- getConversation alice ooConvId >>= getJSON 200
-        shouldBeEmpty $ conv %. "members.others"
+        resp <- getConversation alice ooConvId
+        case resp.status of
+          200 -> do
+            conv <- getJSON 200 resp
+            shouldBeEmpty $ conv %. "members.others"
+          404 -> resp.json %. "label" `shouldMatch` ("no-conversation" :: String)
+          status -> assertFailure $ "Unexpected status while fetching one-to-one conversation: " <> show status
       do
         -- Bob is not in the main conversation any more
         mainConvAfter <- getConversation alice (mainConvBefore %. "qualified_id") >>= getJSON 200

libs/wire-subsystems/src/Wire/ConversationSubsystem/Query.hs

@@ -84,6 +84,7 @@ import Wire.API.Conversation.Role
 import Wire.API.Conversation.Role qualified as Public
 import Wire.API.Error
 import Wire.API.Error.Galley
+import Wire.API.Event.Conversation (SystemEvent (..), SystemEventData (EdSystemConvDelete))
 import Wire.API.Federation.API
 import Wire.API.Federation.API.Galley
 import Wire.API.Federation.Client (FederatorClient, getNegotiatedVersion)
@@ -106,12 +107,16 @@ import Wire.ConversationSubsystem.Fetch (getConversationIdsImpl)
 import Wire.ConversationSubsystem.MLS
 import Wire.ConversationSubsystem.MLS.Enabled (assertMLSEnabled, getMLSPrivateKeys, isMLSEnabled)
 import Wire.ConversationSubsystem.MLS.One2One (localMLSOne2OneConversation, remoteMLSOne2OneConversation)
+import Wire.ConversationSubsystem.Notify qualified as Notify
 import Wire.ConversationSubsystem.One2One
 import Wire.ConversationSubsystem.Util
 import Wire.FeaturesConfigSubsystem
 import Wire.FederationAPIAccess qualified as E
 import Wire.HashPassword (HashPassword)
+import Wire.NotificationSubsystem
 import Wire.RateLimit
+import Wire.Sem.Now (Now)
+import Wire.Sem.Now qualified as Now
 import Wire.Sem.Paging.Cassandra
 import Wire.StoredConversation
 import Wire.StoredConversation qualified as Data
@@ -184,6 +189,8 @@ getConversation ::
     Member (Error FederationError) r,
     Member (E.FederationAPIAccess FederatorClient) r,
     Member P.TinyLog r,
+    Member Now r,
+    Member NotificationSubsystem r,
     Member TeamSubsystem r
   ) =>
   Local UserId ->
@@ -205,6 +212,8 @@ getOwnConversation ::
     Member (Error InternalError) r,
     Member (E.FederationAPIAccess FederatorClient) r,
     Member P.TinyLog r,
+    Member Now r,
+    Member NotificationSubsystem r,
     Member TeamSubsystem r
   ) =>
   Local UserId ->
@@ -222,7 +231,9 @@ getRemoteConversation ::
     Member (ErrorS ConvNotFound) r,
     Member (Error FederationError) r,
     Member TinyLog r,
-    Member (E.FederationAPIAccess FederatorClient) r
+    Member (E.FederationAPIAccess FederatorClient) r,
+    Member Now r,
+    Member NotificationSubsystem r
   ) =>
   Local UserId ->
   Remote ConvId ->
@@ -239,7 +250,9 @@ getRemoteConversations ::
     Member (Error FederationError) r,
     Member (ErrorS 'ConvNotFound) r,
     Member (E.FederationAPIAccess FederatorClient) r,
-    Member P.TinyLog r
+    Member P.TinyLog r,
+    Member Now r,
+    Member NotificationSubsystem r
   ) =>
   Local UserId ->
   [Remote ConvId] ->
@@ -308,7 +321,9 @@ partitionGetConversationFailures = bimap concat concat . partitionEithers . map
 getRemoteConversationsWithFailures ::
   ( Member ConversationStore.ConversationStore r,
     Member (E.FederationAPIAccess FederatorClient) r,
-    Member P.TinyLog r
+    Member P.TinyLog r,
+    Member Now r,
+    Member NotificationSubsystem r
   ) =>
   Local UserId ->
   [Remote ConvId] ->
@@ -346,18 +361,38 @@ getRemoteConversationsWithFailures lusr convs = do
       rpc $ GetConversationsRequest (tUnqualified lusr) (tUnqualified someConvs)
   bimap (localFailures <>) (map remoteView . concat)
     . partitionEithers
-    <$> traverse handleFailure resp
+    <$> traverse (handleRequest locallyFound) resp
   where
-    handleFailure ::
-      (Member P.TinyLog r) =>
+    handleRequest ::
+      ( Member ConversationStore.ConversationStore r,
+        Member P.TinyLog r,
+        Member Now r,
+        Member NotificationSubsystem r
+      ) =>
+      [Remote ConvId] ->
       Either (Remote [ConvId], FederationError) (Remote GetRemoteConversationViewsResponse) ->
       Sem r (Either FailedGetConversation [Remote RemoteConversationView])
-    handleFailure (Left (rcids, e)) = do
+    handleRequest _ (Left (rcids, e)) = do
       P.warn $
         Logger.msg ("Error occurred while fetching remote conversations" :: ByteString)
           . Logger.field "error" (displayException e)
       pure . Left $ failedGetConversationRemotely (sequenceA rcids) e
-    handleFailure (Right c) = pure . Right . traverse (.convs) $ c
+    handleRequest locallyFound (Right response) = do
+      let locallyFoundForDomain = Set.fromList $ filter ((== tDomain response) . tDomain) locallyFound
+          returnedIds = Set.fromList $ map (qualifyAs response . (.id)) (tUnqualified response).convs
+          missingConversations = Set.toList $ locallyFoundForDomain `Set.difference` returnedIds
+      unless (null missingConversations) $ do
+        now <- Now.get
+        for_ missingConversations $ \conv -> do
+          ConversationStore.deleteMembersInRemoteConversation conv [tUnqualified lusr]
+          Notify.pushSystemEvent
+            Nothing
+            (SystemEvent (tUntagged conv) Nothing now Nothing EdSystemConvDelete)
+            (Set.singleton $ tUnqualified lusr)
+        P.info $
+          Logger.msg ("Removed stale local memberships for remote conversations" :: ByteString)
+            . Logger.field "convIds" (show $ map tUntagged missingConversations)
+      pure . Right . traverse (.convs) $ response
 
 getConversationRoles ::
   ( Member ConversationStore.ConversationStore r,
@@ -505,7 +540,9 @@ listConversations ::
   ( Member ConversationStore.ConversationStore r,
     Member (Error InternalError) r,
     Member (E.FederationAPIAccess FederatorClient) r,
-    Member P.TinyLog r
+    Member P.TinyLog r,
+    Member Now r,
+    Member NotificationSubsystem r
   ) =>
   Local UserId ->
   Public.ListConversations ->
@@ -529,9 +566,6 @@ listConversations luser (Public.ListConversations ids) = do
       fetchedOrFailedRemoteIds = Set.fromList $ map Public.cnvQualifiedId remoteConversations <> failedConvs
       remoteNotFoundRemoteIds = filter (`Set.notMember` fetchedOrFailedRemoteIds) $ map tUntagged remoteIds
   unless (null remoteNotFoundRemoteIds) $
-    -- FUTUREWORK: This implies that the backends are out of sync. Maybe the
-    -- current user should be considered removed from this conversation at this
-    -- point.
     P.warn $
       Logger.msg ("Some locally found conversation ids were not returned by remotes" :: ByteString)
         . Logger.field "convIds" (show remoteNotFoundRemoteIds)
@@ -591,7 +625,9 @@ getSelfMember ::
     Member (ErrorS ConvNotFound) r,
     Member (Error FederationError) r,
     Member TinyLog r,
-    Member (E.FederationAPIAccess FederatorClient) r
+    Member (E.FederationAPIAccess FederatorClient) r,
+    Member Now r,
+    Member NotificationSubsystem r
   ) =>
   Local UserId ->
   Qualified ConvId ->