WPB-240: Generate and fan out events about stopping to federate

changelog.d/1-api-changes/WPB-240

@@ -0,0 +1,3 @@
+Added a new notification event type, "federation.delete".
+This event contains a single domain for a remote server that the local server is de-federating from.
+This notification is sent twice during de-federation. Once before and once after cleaning up and removing references to the remote server from the local database.

changelog.d/6-federation/WPB-240

@@ -0,0 +1 @@
+De-federating from a remote server sends a pair of notifications to clients, announcing which server will no longer be federated with.

libs/wire-api/src/Wire/API/Event/Federation.hs

@@ -0,0 +1,64 @@
+module Wire.API.Event.Federation
+  ( Event (..),
+    EventType (..),
+  )
+where
+
+import Data.Aeson (FromJSON, ToJSON)
+import qualified Data.Aeson as A
+import qualified Data.Aeson.KeyMap as KeyMap
+import Data.Domain
+import Data.Json.Util (ToJSONObject (toJSONObject))
+import Data.Schema
+import qualified Data.Swagger as S
+import Imports
+import Wire.Arbitrary
+
+data Event = Event
+  { _eventType :: EventType,
+    _eventDomain :: Domain
+  }
+  deriving (Eq, Show, Ord, Generic)
+
+instance Arbitrary Event where
+  arbitrary =
+    Event
+      <$> arbitrary
+      <*> arbitrary
+
+data EventType
+  = FederationDelete
+  deriving (Eq, Show, Ord, Generic)
+  deriving (Arbitrary) via (GenericUniform EventType)
+  deriving (A.FromJSON, A.ToJSON, S.ToSchema) via Schema EventType
+
+instance ToSchema EventType where
+  schema =
+    enum @Text "EventType" $
+      mconcat
+        [ element "federation.delete" FederationDelete
+        ]
+
+eventObjectSchema :: ObjectSchema SwaggerDoc Event
+eventObjectSchema =
+  Event
+    <$> _eventType .= field "type" schema
+    <*> _eventDomain .= field "domain" schema
+
+instance ToSchema Event where
+  schema = object "Event" eventObjectSchema
+
+instance ToJSONObject Event where
+  toJSONObject =
+    KeyMap.fromList
+      . fromMaybe []
+      . schemaOut eventObjectSchema
+
+instance S.ToSchema Event where
+  declareNamedSchema = schemaToSwagger
+
+instance FromJSON Event where
+  parseJSON = schemaParseJSON
+
+instance ToJSON Event where
+  toJSON = schemaToJSON

libs/wire-api/wire-api.cabal

@@ -37,6 +37,7 @@ library
     Wire.API.Error.Gundeck
     Wire.API.Event.Conversation
     Wire.API.Event.FeatureConfig
+    Wire.API.Event.Federation
     Wire.API.Event.Team
     Wire.API.FederationStatus
     Wire.API.FederationUpdate

services/galley/default.nix

@@ -26,6 +26,7 @@
 , conduit
 , containers
 , cookie
+, cql-io
 , cryptonite
 , currency-codes
 , data-default
@@ -149,6 +150,7 @@ mkDerivation {
     cereal
     comonad
     containers
+    cql-io
     cryptonite
     currency-codes
     data-default

services/galley/galley.cabal

@@ -214,6 +214,7 @@ library
     , cereal                 >=0.4
     , comonad
     , containers             >=0.5
+    , cql-io
     , cryptonite
     , currency-codes         >=2.0
     , data-default           >=0.5

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

@@ -26,6 +26,7 @@ module Galley.API.Internal
   )
 where
 
+import Cassandra (ClientState, Consistency (LocalQuorum), Page (..), paramsP)
 import Control.Exception
 import Control.Exception.Safe (catchAny)
 import Control.Lens hiding (Getter, Setter, (.=))
@@ -39,6 +40,7 @@ import Data.Range
 import Data.Singletons
 import Data.Text (unpack)
 import Data.Time
+import Database.CQL.IO (paginate)
 import Galley.API.Action
 import qualified Galley.API.Clients as Clients
 import qualified Galley.API.Create as Create
@@ -59,12 +61,16 @@ import Galley.API.Teams.Features
 import qualified Galley.API.Update as Update
 import Galley.API.Util
 import Galley.App
+import Galley.Cassandra.Conversation.Members
+import Galley.Cassandra.Queries
+import Galley.Cassandra.Store (embedClient)
 import qualified Galley.Data.Conversation as Data
 import Galley.Data.Conversation.Types
 import Galley.Effects
 import Galley.Effects.BackendNotificationQueueAccess
 import Galley.Effects.ClientStore
 import Galley.Effects.ConversationStore
+import Galley.Effects.ExternalAccess
 import Galley.Effects.FederatorAccess
 import Galley.Effects.GundeckAccess
 import Galley.Effects.LegalHoldStore as LegalHoldStore
@@ -73,7 +79,9 @@ import qualified Galley.Effects.MemberStore as E
 import Galley.Effects.ProposalStore
 import Galley.Effects.TeamStore
 import qualified Galley.Effects.TeamStore as E
+import Galley.Env (currentFanoutLimit, _options)
 import qualified Galley.Intra.Push as Intra
+import Galley.Intra.Push.Internal (pushEventJson)
 import Galley.Monad
 import Galley.Options
 import qualified Galley.Queue as Q
@@ -85,8 +93,8 @@ import Imports hiding (head)
 import qualified Network.AMQP as Q
 import Network.HTTP.Types
 import Network.Wai
-import Network.Wai.Predicate hiding (Error, err, setStatus)
-import qualified Network.Wai.Predicate as Predicate
+import Network.Wai.Predicate hiding (Error, err, result, setStatus)
+import qualified Network.Wai.Predicate as Predicate hiding (result)
 import Network.Wai.Routing hiding (App, route, toList)
 import Network.Wai.Utilities hiding (Error)
 import Network.Wai.Utilities.ZAuth
@@ -106,6 +114,7 @@ import Wire.API.CustomBackend
 import Wire.API.Error
 import Wire.API.Error.Galley
 import Wire.API.Event.Conversation
+import qualified Wire.API.Event.Federation as Federation
 import Wire.API.Federation.API
 import Wire.API.Federation.API.Galley
 import qualified Wire.API.Federation.API.Galley as F
@@ -534,6 +543,7 @@ internalDeleteFederationDomainH ::
     Member MemberStore r,
     Member ConversationStore r,
     Member (Embed IO) r,
+    Member (Input ClientState) r,
     Member CodeStore r,
     Member TeamStore r,
     Member BrigAccess r,
@@ -543,8 +553,35 @@ internalDeleteFederationDomainH ::
   Domain ::: JSON ->
   Sem r Response
 internalDeleteFederationDomainH (domain ::: _) = do
+  -- We have to send the same event twice.
+  -- Once before and once after defederation work.
+  -- https://wearezeta.atlassian.net/wiki/spaces/ENGINEERIN/pages/809238539/Use+case+Stopping+to+federate+with+a+domain
+  void sendNotifications
   deleteFederationDomain domain
+  void sendNotifications
   pure (empty & setStatus status200)
+  where
+    sendNotifications = do
+      maxPage <- inputs $ fromRange . currentFanoutLimit . _options -- This is based on the limits in removeIfLargeFanout
+      page <- embedClient $ paginate selectAllMembers (paramsP LocalQuorum () maxPage)
+      sendNotificationPage page
+    pushEvents results = do
+      let (bots, mems) = localBotsAndUsers results
+          recipients = Intra.recipient <$> mems
+          event = Intra.FederationEvent $ Federation.Event Federation.FederationDelete domain
+      for_ (Intra.newPush ListComplete Nothing event recipients) $ \p -> do
+        -- TODO: Transient or not?
+        -- RouteAny is used as it will wake up mobile clients
+        -- and notify them of the changes to federation state.
+        push1 $ p & Intra.pushRoute .~ Intra.RouteAny
+      deliverAsync (bots `zip` repeat (pushEventJson event))
+    sendNotificationPage page = do
+      let res = result page
+          mems = mapMaybe toMember res
+      pushEvents mems
+      when (hasMore page) $ do
+        page' <- embedClient $ nextPage page
+        sendNotificationPage page'
 
 -- Remove remote members from local conversations
 deleteFederationDomainRemoteUserFromLocalConversations ::

services/galley/src/Galley/Cassandra/Conversation/Members.hs

@@ -18,6 +18,8 @@
 module Galley.Cassandra.Conversation.Members
   ( addMembers,
     members,
+    allMembers,
+    toMember,
     lookupRemoteMembers,
     removeMembersFromLocalConv,
     toMemberStatus,
@@ -122,6 +124,11 @@ members conv =
   fmap (mapMaybe toMember) . retry x1 $
     query Cql.selectMembers (params LocalQuorum (Identity conv))
 
+allMembers :: Client [LocalMember]
+allMembers =
+  fmap (mapMaybe toMember) . retry x1 $
+    query Cql.selectAllMembers (params LocalQuorum ())
+
 toMemberStatus ::
   ( -- otr muted
     Maybe MutedStatus,
@@ -386,6 +393,7 @@ interpretMemberStoreToCassandra = interpret $ \case
   CreateBotMember sr bid cid -> embedClient $ addBotMember sr bid cid
   GetLocalMember cid uid -> embedClient $ member cid uid
   GetLocalMembers cid -> embedClient $ members cid
+  GetAllLocalMembers -> embedClient allMembers
   GetRemoteMember cid uid -> embedClient $ lookupRemoteMember cid (tDomain uid) (tUnqualified uid)
   GetRemoteMembers rcid -> embedClient $ lookupRemoteMembers rcid
   CheckLocalMemberRemoteConv uid rcnv -> fmap (not . null) $ embedClient $ lookupLocalMemberRemoteConv uid rcnv

services/galley/src/Galley/Cassandra/Queries.hs

@@ -339,6 +339,9 @@ selectMember = "select user, service, provider, status, otr_muted_status, otr_mu
 selectMembers :: PrepQuery R (Identity ConvId) (UserId, Maybe ServiceId, Maybe ProviderId, Maybe MemberStatus, Maybe MutedStatus, Maybe Text, Maybe Bool, Maybe Text, Maybe Bool, Maybe Text, Maybe RoleName)
 selectMembers = "select user, service, provider, status, otr_muted_status, otr_muted_ref, otr_archived, otr_archived_ref, hidden, hidden_ref, conversation_role from member where conv = ?"
 
+selectAllMembers :: PrepQuery R () (UserId, Maybe ServiceId, Maybe ProviderId, Maybe MemberStatus, Maybe MutedStatus, Maybe Text, Maybe Bool, Maybe Text, Maybe Bool, Maybe Text, Maybe RoleName)
+selectAllMembers = "select user, service, provider, status, otr_muted_status, otr_muted_ref, otr_archived, otr_archived_ref, hidden, hidden_ref, conversation_role from member"
+
 insertMember :: PrepQuery W (ConvId, UserId, Maybe ServiceId, Maybe ProviderId, RoleName) ()
 insertMember = "insert into member (conv, user, service, provider, status, conversation_role) values (?, ?, ?, ?, 0, ?)"
 

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

@@ -26,6 +26,7 @@ module Galley.Effects.ExternalAccess
   )
 where
 
+import Data.Aeson
 import Data.Id
 import Galley.Data.Services
 import Imports
@@ -34,7 +35,7 @@ import Wire.API.Event.Conversation
 
 data ExternalAccess m a where
   Deliver :: Foldable f => f (BotMember, Event) -> ExternalAccess m [BotMember]
-  DeliverAsync :: Foldable f => f (BotMember, Event) -> ExternalAccess m ()
+  DeliverAsync :: (ToJSON e, Foldable f) => f (BotMember, e) -> ExternalAccess m ()
   DeliverAndDeleteAsync :: Foldable f => ConvId -> f (BotMember, Event) -> ExternalAccess m ()
 
 makeSem ''ExternalAccess

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

@@ -30,6 +30,7 @@ module Galley.Effects.MemberStore
     -- * Read members
     getLocalMember,
     getLocalMembers,
+    getAllLocalMembers,
     getRemoteMember,
     getRemoteMembers,
     checkLocalMemberRemoteConv,
@@ -70,6 +71,7 @@ data MemberStore m a where
   CreateBotMember :: ServiceRef -> BotId -> ConvId -> MemberStore m BotMember
   GetLocalMember :: ConvId -> UserId -> MemberStore m (Maybe LocalMember)
   GetLocalMembers :: ConvId -> MemberStore m [LocalMember]
+  GetAllLocalMembers :: MemberStore m [LocalMember]
   GetRemoteMember :: ConvId -> Remote UserId -> MemberStore m (Maybe RemoteMember)
   GetRemoteMembers :: ConvId -> MemberStore m [RemoteMember]
   CheckLocalMemberRemoteConv :: UserId -> Remote ConvId -> MemberStore m Bool

services/galley/src/Galley/External.hs

@@ -21,6 +21,7 @@ import Bilge.Request
 import Bilge.Retry (httpHandlers)
 import Control.Lens
 import Control.Retry
+import Data.Aeson (ToJSON)
 import Data.ByteString.Conversion.To
 import Data.Id
 import Data.Misc
@@ -60,7 +61,7 @@ interpretExternalAccess = interpret $ \case
 -- | Like deliver, but ignore orphaned bots and return immediately.
 --
 -- FUTUREWORK: Check if this can be removed.
-deliverAsync :: [(BotMember, Event)] -> App ()
+deliverAsync :: ToJSON e => [(BotMember, e)] -> App ()
 deliverAsync = void . forkIO . void . deliver
 
 -- | Like deliver, but remove orphaned bots and return immediately.
@@ -69,10 +70,10 @@ deliverAndDeleteAsync cnv pushes = void . forkIO $ do
   gone <- deliver pushes
   mapM_ (deleteBot cnv . botMemId) gone
 
-deliver :: [(BotMember, Event)] -> App [BotMember]
+deliver :: forall e. ToJSON e => [(BotMember, e)] -> App [BotMember]
 deliver pp = mapM (async . exec) pp >>= foldM eval [] . zip (map fst pp)
   where
-    exec :: (BotMember, Event) -> App Bool
+    exec :: (BotMember, e) -> App Bool
     exec (b, e) =
       lookupService (botMemService b) >>= \case
         Nothing -> pure False
@@ -118,7 +119,7 @@ deliver pp = mapM (async . exec) pp >>= foldM eval [] . zip (map fst pp)
 
 -- Internal -------------------------------------------------------------------
 
-deliver1 :: Service -> BotMember -> Event -> App ()
+deliver1 :: ToJSON e => Service -> BotMember -> e -> App ()
 deliver1 s bm e
   | s ^. serviceEnabled = do
       let t = toByteString' (s ^. serviceToken)

services/galley/src/Galley/Intra/Push/Internal.hs

@@ -40,18 +40,21 @@ import Imports hiding (forkIO)
 import UnliftIO.Async (mapConcurrently_)
 import Wire.API.Event.Conversation (Event (evtFrom))
 import qualified Wire.API.Event.FeatureConfig as FeatureConfig
+import qualified Wire.API.Event.Federation as Federation
 import qualified Wire.API.Event.Team as Teams
 import Wire.API.Team.Member
 
 data PushEvent
   = ConvEvent Event
   | TeamEvent Teams.Event
   | FeatureConfigEvent FeatureConfig.Event
+  | FederationEvent Federation.Event
 
 pushEventJson :: PushEvent -> Object
 pushEventJson (ConvEvent e) = toJSONObject e
 pushEventJson (TeamEvent e) = toJSONObject e
 pushEventJson (FeatureConfigEvent e) = toJSONObject e
+pushEventJson (FederationEvent e) = toJSONObject e
 
 data RecipientBy user = Recipient
   { _recipientUserId :: user,