Send notifications using Rabbitmq instead of Cassandra

charts/cannon/templates/configmap.yaml

@@ -14,6 +14,19 @@ data:
       host: gundeck
       port: 8080
 
+    {{- with .Values.config.rabbitmq }}
+    rabbitmq:
+      host: {{ .host }}
+      adminPort: {{ .adminPort }}
+      port: {{ .port }}
+      vHost: {{ .vHost }}
+      enableTls: {{ .enableTls }}
+      insecureSkipVerifyTls: {{ .insecureSkipVerifyTls }}
+      {{- if .tlsCaSecretRef }}
+      caCert: /etc/wire/cannon/rabbitmq-ca/{{ .tlsCaSecretRef.key }}
+      {{- end }}
+    {{- end }}
+
     drainOpts:
       gracePeriodSeconds: {{ .Values.config.drainOpts.gracePeriodSeconds }}
       millisecondsBetweenBatches: {{ .Values.config.drainOpts.millisecondsBetweenBatches }}

charts/cannon/templates/statefulset.yaml

@@ -95,6 +95,10 @@ spec:
           mountPath: /etc/wire/cannon/externalHost
         - name: cannon-config
           mountPath: /etc/wire/cannon/conf
+        {{- if .Values.config.rabbitmq.tlsCaSecretRef }}
+        - name: "rabbitmq-ca"
+          mountPath: "/etc/wire/cannon/rabbitmq-ca/"
+        {{- end }}
         ports:
         - name: http
           containerPort: {{ .Values.service.internalPort }}
@@ -148,3 +152,8 @@ spec:
         secret:
           secretName: {{ .Values.service.nginz.tls.secretName }}
       {{- end }}
+      {{- if .Values.config.rabbitmq.tlsCaSecretRef }}
+      - name: "rabbitmq-ca"
+        secret:
+          secretName: {{ .Values.config.rabbitmq.tlsCaSecretRef.name }}
+      {{- end }}

charts/cannon/values.yaml

@@ -11,6 +11,16 @@ config:
   logLevel: Info
   logFormat: StructuredJSON
   logNetStrings: false
+  rabbitmq:
+    host: rabbitmq
+    port: 5672
+    adminPort: 15672
+    vHost: /
+    enableTls: false
+    insecureSkipVerifyTls: false
+    # tlsCaSecretRef:
+    #   name: <secret-name>
+    #   key: <ca-attribute>
 
   # See also the section 'Controlling the speed of websocket draining during
   # cannon pod replacement' in docs/how-to/install/configuration-options.rst

charts/gundeck/templates/configmap.yaml

@@ -29,25 +29,15 @@ data:
       tlsCa: /etc/wire/gundeck/cassandra/{{- (include "tlsSecretRef" . | fromYaml).key }}
       {{- end }}
 
-    redis:
-      host: {{ .redis.host }}
-      port: {{ .redis.port }}
-      connectionMode: {{ .redis.connectionMode }}
-      enableTls: {{ .redis.enableTls }}
-      insecureSkipVerifyTls: {{ .redis.insecureSkipVerifyTls }}
-      {{- if eq (include "configureRedisCa" .) "true" }}
-      tlsCa: /etc/wire/gundeck/redis-ca/{{ include "redisTlsSecretKey" .}}
-      {{- end }}
-
-    {{- if .redisAdditionalWrite }}
-    redisAdditionalWrite:
-      host: {{ .redisAdditionalWrite.host }}
-      port: {{ .redisAdditionalWrite.port }}
-      connectionMode: {{ .redisAdditionalWrite.connectionMode }}
-      enableTls: {{ .redisAdditionalWrite.enableTls }}
-      insecureSkipVerifyTls: {{ .redisAdditionalWrite.insecureSkipVerifyTls }}
-      {{- if eq (include "configureAdditionalRedisCa" .) "true" }}
-      tlsCa: /etc/wire/gundeck/additional-redis-ca/{{ include "additionalRedisTlsSecretKey" .}}
+    {{- with .rabbitmq }}
+    rabbitmq:
+      host: {{ .host }}
+      port: {{ .port }}
+      vHost: {{ .vHost }}
+      enableTls: {{ .enableTls }}
+      insecureSkipVerifyTls: {{ .insecureSkipVerifyTls }}
+      {{- if .tlsCaSecretRef }}
+      caCert: /etc/wire/gundeck/rabbitmq-ca/{{ .tlsCaSecretRef.key }}
       {{- end }}
     {{- end }}
 

charts/gundeck/templates/deployment.yaml

@@ -37,15 +37,10 @@ spec:
           secret:
             secretName: {{ (include "tlsSecretRef" .Values.config | fromYaml).name }}
         {{- end}}
-        {{- if eq (include "configureRedisCa" .Values.config) "true" }}
-        - name: "redis-ca"
+        {{- if .Values.config.rabbitmq.tlsCaSecretRef }}
+        - name: "rabbitmq-ca"
           secret:
-            secretName: {{ include "redisTlsSecretName" .Values.config }}
-        {{- end }}
-        {{- if eq (include "configureAdditionalRedisCa" .Values.config) "true" }}
-        - name: "additional-redis-ca"
-          secret:
-            secretName: {{ include "additionalRedisTlsSecretName" .Values.config }}
+            secretName: {{ .Values.config.rabbitmq.tlsCaSecretRef.name }}
         {{- end }}
       containers:
         - name: gundeck
@@ -62,13 +57,9 @@ spec:
           - name: "gundeck-cassandra"
             mountPath: "/etc/wire/gundeck/cassandra"
           {{- end }}
-          {{- if eq (include "configureRedisCa" .Values.config) "true" }}
-          - name: "redis-ca"
-            mountPath: "/etc/wire/gundeck/redis-ca/"
-          {{- end }}
-          {{- if eq (include "configureAdditionalRedisCa" .Values.config) "true" }}
-          - name: "additional-redis-ca"
-            mountPath: "/etc/wire/gundeck/additional-redis-ca/"
+          {{- if .Values.config.rabbitmq.tlsCaSecretRef }}
+          - name: "rabbitmq-ca"
+            mountPath: "/etc/wire/gundeck/rabbitmq-ca/"
           {{- end }}
           env:
           {{- if hasKey .Values.secrets "awsKeyId" }}

charts/gundeck/values.yaml

@@ -27,35 +27,15 @@ config:
     # tlsCaSecretRef:
     #   name: <secret-name>
     #   key: <ca-attribute>
-  redis:
-    host: redis-ephemeral-master
-    port: 6379
-    connectionMode: "master" # master | cluster
+  rabbitmq:
+    host: rabbitmq
+    port: 5672
+    vHost: /
     enableTls: false
     insecureSkipVerifyTls: false
-    # To configure custom TLS CA, please provide one of these:
-    # tlsCa: <CA in PEM format (can be self-signed)>
-    #
-    # Or refer to an existing secret (containing the CA):
     # tlsCaSecretRef:
     #   name: <secret-name>
     #   key: <ca-attribute>
-
-  # To enable additional writes during a migration:
-  # redisAdditionalWrite:
-  #   host: redis-two
-  #   port: 6379
-  #   connectionMode: master
-  #   enableTls: false
-  #   insecureSkipVerifyTls: false
-  #
-  # # To configure custom TLS CA, please provide one of these:
-  # # tlsCa: <CA in PEM format (can be self-signed)>
-  # #
-  # # Or refer to an existing secret (containing the CA):
-  # # tlsCaSecretRef:
-  # #   name: <secret-name>
-  # #   key: <ca-attribute>
   bulkPush: true
   aws:
     region: "eu-west-1"

charts/rabbitmq/values.yaml

@@ -0,0 +1,2 @@
+rabbitmq:
+  extraPlugins: rabbitmq_stream

deploy/dockerephemeral/docker-compose.yaml

@@ -262,7 +262,11 @@ services:
 
   rabbitmq:
     container_name: rabbitmq
-    image: rabbitmq:3.11-management-alpine
+    build:
+      context: .
+      dockerfile_inline: |
+        FROM rabbitmq:3.11-management-alpine
+        RUN rabbitmq-plugins enable rabbitmq_stream
     environment:
       - RABBITMQ_USERNAME
       - RABBITMQ_PASSWORD

integration/test/Test/MLS/Message.hs

@@ -21,11 +21,33 @@ module Test.MLS.Message where
 
 import API.Galley
 import API.Gundeck
+import qualified Data.Aeson as Aeson
 import MLS.Util
 import Notifications
 import SetupHelpers
 import Testlib.Prelude
 
+-- import UnliftIO.Concurrent (threadDelay)
+
+testFoo :: (HasCallStack) => App ()
+testFoo = replicateM_ 10 $ do
+  alice <- randomUser OwnDomain def
+  printJSON alice
+  -- threadDelay 1000000
+  withWebSocket alice $ \ws -> do
+    -- liftIO $ threadDelay 1000000
+    void $ createMLSClient def alice
+    n <- awaitMatch isUserClientAddNotif ws
+    printJSON n
+
+testBar :: (HasCallStack) => App ()
+testBar = do
+  alice <- randomUser OwnDomain def
+  printJSON alice
+  getNotifications alice def `bindResponse` \resp -> do
+    resp.status `shouldMatchInt` 200
+    resp.json `shouldMatch` Aeson.Null
+
 -- | Test happy case of federated MLS message sending in both directions.
 testApplicationMessage :: (HasCallStack) => App ()
 testApplicationMessage = do

integration/test/Testlib/Cannon.hs

@@ -68,7 +68,6 @@ import qualified Network.HTTP.Client as Http
 import qualified Network.WebSockets as WS
 import System.Random (randomIO)
 import System.Timeout (timeout)
-import Testlib.App
 import Testlib.Assertions
 import Testlib.Env
 import Testlib.HTTP
@@ -148,9 +147,11 @@ clientApp wsChan latch conn = do
       case decodeStrict' bs of
         Just n -> atomically $ writeTChan wsChan n
         Nothing -> putStrLn $ "Failed to decode notification: " ++ show bs
-    wsWrite = forever $ do
-      takeMVar latch
-      WS.sendClose conn ("close" :: ByteString)
+    wsWrite = do
+      WS.sendPing conn ("hello" :: ByteString)
+      forever $ do
+        takeMVar latch
+        WS.sendClose conn ("close" :: ByteString)
 
 -- | Start a client thread in 'Async' that opens a web socket to a Cannon, wait
 --   for the connection to register with Gundeck, and return the 'Async' thread.
@@ -164,7 +165,6 @@ run wsConnect app = do
   serviceMap <- getServiceMap domain
 
   let HostPort caHost caPort = serviceHostPort serviceMap Cannon
-  latch <- liftIO newEmptyMVar
 
   connId <- case wsConnect.conn of
     Just c -> pure c
@@ -184,6 +184,7 @@ run wsConnect app = do
     r <- rawBaseRequest domain Cannon Versioned path
     pure r {HTTP.requestHeaders = caHdrs}
 
+  waitForPong <- liftIO $ newEmptyMVar
   wsapp <-
     liftIO
       $ async
@@ -192,21 +193,11 @@ run wsConnect app = do
             caHost
             (fromIntegral caPort)
             path
-            WS.defaultConnectionOptions
+            (WS.defaultConnectionOptions {WS.connectionOnPong = void $ tryPutMVar waitForPong ()})
             caHdrs
             app
         )
-      $ \(e :: SomeException) -> putMVar latch e
-
-  presenceRequest <-
-    baseRequest domain Cannon Unversioned $
-      "/i/presences/" <> wsConnect.user <> "/" <> connId
-
-  waitForPresence <- appToIO $ retryT $ do
-    response <- submit "HEAD" presenceRequest
-    status response `shouldMatchInt` 200
-  let waitForException = do
-        ex <- takeMVar latch
+      $ \(ex :: SomeException) -> do
         -- Construct a "fake" response. We do not really have access to the
         -- websocket connection requests and response, unfortunately, but it is
         -- useful to display some information about the request in case an
@@ -220,8 +211,8 @@ run wsConnect app = do
                   request = request
                 }
         throwIO (AssertionFailure callStack (Just r) (displayException ex))
-
-  liftIO $ race_ waitForPresence waitForException
+  -- TODO: add a race so we timeout
+  liftIO $ takeMVar waitForPong
   pure wsapp
 
 close :: (MonadIO m) => WebSocket -> m ()

libs/extended/src/Network/AMQP/Extended.hs

@@ -55,7 +55,7 @@ data RabbitMqTlsOpts = RabbitMqTlsOpts
   { caCert :: !(Maybe FilePath),
     insecureSkipVerifyTls :: Bool
   }
-  deriving (Show)
+  deriving (Show, Eq)
 
 parseTlsJson :: Object -> Parser (Maybe RabbitMqTlsOpts)
 parseTlsJson v = do
@@ -76,7 +76,7 @@ data RabbitMqAdminOpts = RabbitMqAdminOpts
     tls :: Maybe RabbitMqTlsOpts,
     adminPort :: !Int
   }
-  deriving (Show)
+  deriving (Eq, Show)
 
 instance FromJSON RabbitMqAdminOpts where
   parseJSON = withObject "RabbitMqAdminOpts" $ \v ->
@@ -87,6 +87,9 @@ instance FromJSON RabbitMqAdminOpts where
       <*> parseTlsJson v
       <*> v .: "adminPort"
 
+instance ToJSON RabbitMqAdminOpts where
+  toJSON = error "RabbitMqAdminOpts toJSON not implemented due to developer laziness"
+
 mkRabbitMqAdminClientEnv :: RabbitMqAdminOpts -> IO (AdminAPI (AsClientT IO))
 mkRabbitMqAdminClientEnv opts = do
   (username, password) <- readCredsFromEnv
@@ -111,7 +114,7 @@ data RabbitMqOpts = RabbitMqOpts
     vHost :: !Text,
     tls :: !(Maybe RabbitMqTlsOpts)
   }
-  deriving (Show)
+  deriving (Show, Eq)
 
 instance FromJSON RabbitMqOpts where
   parseJSON = withObject "RabbitMqAdminOpts" $ \v ->
@@ -121,6 +124,9 @@ instance FromJSON RabbitMqOpts where
       <*> v .: "vHost"
       <*> parseTlsJson v
 
+instance ToJSON RabbitMqOpts where
+  toJSON = error "RabbitMqOpts toJSON not implemented due to developer laziness"
+
 demoteOpts :: RabbitMqAdminOpts -> RabbitMqOpts
 demoteOpts RabbitMqAdminOpts {..} = RabbitMqOpts {..}
 

libs/extended/src/Network/RabbitMqAdmin.hs

@@ -25,6 +25,13 @@ data AdminAPI route = AdminAPI
           :> "queues"
           :> Capture "vhost" VHost
           :> Get '[JSON] [Queue],
+    getQueue ::
+      route
+        :- "api"
+          :> "queues"
+          :> Capture "vhost" VHost
+          :> Capture "queue" Text
+          :> Get '[JSON] Queue,
     deleteQueue ::
       route
         :- "api"
@@ -43,7 +50,7 @@ data AuthenticatedAPI route = AuthenticatedAPI
   }
   deriving (Generic)
 
-data Queue = Queue {name :: Text, vhost :: Text}
+data Queue = Queue {name :: Text, vhost :: Text, status :: Maybe Text}
   deriving (Show, Eq, Generic)
 
 instance FromJSON Queue

libs/wire-api/src/Wire/API/Internal/Notification.hs

@@ -54,8 +54,7 @@ import Wire.API.Notification
 -- Notification
 
 data Notification = Notification
-  { ntfId :: !NotificationId,
-    ntfTransient :: !Bool,
+  { ntfTransient :: !Bool,
     ntfPayload :: !(List1 Object)
   }
   deriving (Eq, Show)
@@ -65,8 +64,7 @@ instance S.ToSchema Notification where
   schema =
     S.object "Notification" $
       Notification
-        <$> ntfId S..= S.field "id" S.schema
-        <*> ntfTransient S..= (fromMaybe False <$> S.optField "transient" S.schema)
+        <$> ntfTransient S..= (fromMaybe False <$> S.optField "transient" S.schema)
         <*> (toNonEmpty . ntfPayload) S..= fmap List1 (S.field "payload" (S.nonEmptyArray S.jsonObject))
 
 --------------------------------------------------------------------------------