[DO NOT MERGE] user nofifications pulsar hs

Makefile

@@ -18,7 +18,8 @@ fake-aws fake-aws-s3 fake-aws-sqs aws-ingress fluent-bit kibana backoffice		\
 calling-test demo-smtp elasticsearch-curator elasticsearch-external				\
 elasticsearch-ephemeral minio-external cassandra-external						\
 ingress-nginx-controller nginx-ingress-services reaper restund \
-k8ssandra-test-cluster ldap-scim-bridge wire-server-enterprise
+k8ssandra-test-cluster ldap-scim-bridge wire-server-enterprise \
+integration
 KIND_CLUSTER_NAME     := wire-server
 HELM_PARALLELISM      ?= 1 # 1 for sequential tests; 6 for all-parallel tests
 # (run `psql -h localhost -p 5432 -d backendA -U wire-server -w` for the list of options for PSQL_DB)

charts/cannon/templates/configmap.yaml

@@ -39,6 +39,12 @@ data:
     rabbitMqMaxConnections: {{ .config.rabbitMqMaxConnections }}
     rabbitMqMaxChannels: {{ .config.rabbitMqMaxChannels }}
 
+    {{- with .config.pulsar }}
+    pulsar:
+      host: {{ .host }}
+      port: {{ .port }}
+    {{- end }}
+
     drainOpts:
       gracePeriodSeconds: {{ .config.drainOpts.gracePeriodSeconds }}
       millisecondsBetweenBatches: {{ .config.drainOpts.millisecondsBetweenBatches }}

charts/cannon/values.yaml

@@ -29,6 +29,10 @@ config:
     #   name: <secret-name>
     #   key: <ca-attribute>
 
+  pulsar:
+    host: localhost
+    port: 6650
+
   # See also the section 'Controlling the speed of websocket draining during
   # cannon pod replacement' in docs/how-to/install/configuration-options.rst
   drainOpts:

charts/gundeck/templates/configmap.yaml

@@ -41,6 +41,18 @@ data:
       {{- end }}
     {{- end }}
 
+    {{- with .pulsar }}
+    pulsar:
+      host: {{ .host }}
+      port: {{ .port }}
+    {{- end }}
+
+    {{- with .pulsarAdmin }}
+    pulsarAdmin:
+      host: {{ .host }}
+      port: {{ .port }}
+    {{- end }}
+
     redis:
       host: {{ .redis.host }}
       port: {{ .redis.port }}

charts/gundeck/values.yaml

@@ -50,6 +50,14 @@ config:
     #   name: <secret-name>
     #   key: <ca-attribute>
 
+  pulsar:
+    host: localhost
+    port: 6650
+
+  pulsarAdmin:
+    host: localhost
+    port: 8080
+
   # To enable additional writes during a migration:
   # redisAdditionalWrite:
   #   host: redis-two

deploy/dockerephemeral/docker-compose.yaml

@@ -419,6 +419,50 @@ services:
       # - DNS_SERVER_RECURSION_ALLOWED_NETWORKS=127.0.0.1, 192.168.1.0/24 #Comma separated list of IP addresses or network addresses to allow recursion. Valid only for `UseSpecifiedNetworkACL` recursion option.  This option is obsolete and DNS_SERVER_RECURSION_NETWORK_ACL should be used instead.
       # - DNS_SERVER_ENABLE_BLOCKING=false #Sets the DNS server to block domain names using Blocked Zone and Block List Zone.
 
+  pulsar:
+    image: apachepulsar/pulsar:latest
+    container_name: pulsar
+    # 8080 belongs to nginz
+    ports:
+      - 6650:6650
+      - 5080:8080
+    networks:
+      - demo_wire
+    command: bin/pulsar standalone
+
+  # Admin GUI
+  # Url: http://localhost:9527
+  # Username: pulsar
+  # Password: walter-frosch
+  pulsar-manager:
+    image: apachepulsar/pulsar-manager:latest
+    container_name: pulsar-manager
+    ports:
+      - 9527:9527
+      - 7750:7750
+    networks:
+      - demo_wire
+    environment:
+      - "SPRING_CONFIGURATION_FILE=/pulsar-manager/pulsar-manager/application.properties"
+    healthcheck:
+      test:
+        ["CMD-SHELL", "curl http://127.0.0.1:7750/actuator/health || exit 1 "]
+      interval: 3s
+      timeout: 5s
+      retries: 100
+
+  pulsar-init:
+    container_name: pulsar-init
+    image: curlimages/curl:latest
+    networks:
+      - demo_wire
+    command: init-pulsar
+    volumes:
+      - ./pulsar-config/init-pulsar.sh:/bin/init-pulsar
+    depends_on:
+      pulsar-manager:
+        condition: service_healthy
+
 volumes:
   redis-node-1-data:
   redis-node-2-data:

deploy/dockerephemeral/pulsar-config/init-pulsar.sh

@@ -0,0 +1,20 @@
+#!/usr/bin/env sh
+
+# Create a superuser in pulsar-manager.
+# username: pulsar
+# password: walter-frosch
+CSRF_TOKEN=$(curl http://pulsar-manager:7750/pulsar-manager/csrf-token)
+curl -v -H "X-XSRF-TOKEN: $CSRF_TOKEN" \
+  -H "Cookie: XSRF-TOKEN=$CSRF_TOKEN;" \
+  -H "Content-Type: application/json" \
+  -X PUT http://pulsar-manager:7750/pulsar-manager/users/superuser \
+  -d '{"name": "pulsar", "password": "walter-frosch", "description": "test", "email": "username@test.org"}'
+
+curl -v -X PUT http://pulsar:8080/admin/v2/tenants/wire \
+ -H "Content-Type: application/json" \
+ -d '{"adminRoles": ["pulsar"], "allowedClusters": ["standalone"]}'
+
+curl -v -X PUT \
+  http://pulsar:8080/admin/v2/namespaces/wire/user-notifications \
+  -H "Content-Type: application/json" \
+  -d '{"message_ttl_in_seconds":3600}'

integration/default.nix

@@ -74,6 +74,7 @@
 , split
 , stm
 , streaming-commons
+, streamly
 , string-conversions
 , system-linux-proc
 , tagged
@@ -179,6 +180,7 @@ mkDerivation {
     split
     stm
     streaming-commons
+    streamly
     string-conversions
     system-linux-proc
     tagged

integration/integration.cabal

@@ -184,6 +184,7 @@ library
     Test.MLS.Unreachable
     Test.NginxZAuthModule
     Test.Notifications
+    Test.NotificationsBenchmark
     Test.OAuth
     Test.PasswordReset
     Test.Presence
@@ -299,6 +300,7 @@ library
     , split
     , stm
     , streaming-commons
+    , streamly
     , string-conversions
     , system-linux-proc
     , tagged

integration/test/Test/NotificationsBenchmark.hs

@@ -0,0 +1,180 @@
+module Test.NotificationsBenchmark where
+
+import API.Brig
+import API.BrigCommon
+import API.Common
+import API.GundeckInternal
+import Control.Concurrent
+import Control.Concurrent.Async (Async)
+import Control.Concurrent.STM.TBQueue
+import Control.Monad.Codensity (Codensity (..))
+import Control.Monad.Reader (MonadReader (ask), asks)
+import Control.Monad.Reader.Class (local)
+import Control.Retry
+import qualified Data.Map.Strict as Map
+import Data.Time
+import Debug.Trace
+import GHC.Conc.Sync
+import GHC.Stack
+import SetupHelpers
+import qualified Streamly.Data.Fold.Prelude as Fold
+import qualified Streamly.Data.Stream.Prelude as Stream
+import System.Random
+import qualified Test.Events as TestEvents
+import Testlib.Prekeys
+import Testlib.Prelude
+import UnliftIO (async, waitAnyCancel)
+
+data TestRecipient = TestRecipient
+  { user :: Value,
+    clientIds :: [String]
+  }
+  deriving (Show)
+
+testBench :: (HasCallStack) => App ()
+testBench = do
+  shardingGroupCount <- asks (.shardingGroupCount)
+  shardingGroup <- asks (.shardingGroup)
+  maxUserNo <- asks (.maxUserNo)
+  env <- ask
+
+  -- Preparation
+  let threadCount = min numCapabilities (fromIntegral maxUserNo)
+      parCfg = Stream.maxThreads threadCount . Stream.ordered False
+      toMap = Fold.foldl' (\kv (k, v) -> Map.insert k v kv) Map.empty
+  -- Later, we only read from this map. Thus, it doesn't have to be thread-safe.
+  userMap :: Map Word TestRecipient <-
+    Stream.fromList [0 :: Word .. maxUserNo]
+      & Stream.filter ((shardingGroup ==) . (`mod` shardingGroupCount))
+      & Stream.parMapM parCfg (\i -> generateTestRecipient >>= \r -> pure (i, r))
+      & Stream.fold toMap
+
+  now <- liftIO getCurrentTime
+
+  -- TODO: To be replaced with real data from the file. (See
+  -- https://wearezeta.atlassian.net/wiki/spaces/PET/pages/2118680620/Simulating+production-like+data)
+  let fakeData = zip (plusDelta now <$> [0 :: Word ..]) (cycle [1 .. maxUserNo])
+
+  tbchans :: [(Int, TBQueue (UTCTime, Word))] <- liftIO $ forM [1 .. threadCount] $ \i -> do
+    q <- atomically $ newTBQueue 1 -- capacity 100, adjust as needed
+    pure (i, q)
+
+  workers :: [Async ()] <- forM tbchans $ \(i, chan) -> liftIO
+    . async
+    -- . handle (\(e :: SomeException) -> traceM $ "Caught exception in worker id=" <> show i <> " e=" <> show e)
+    . forever
+    . runAppWithEnv env
+    $ do
+      traceM $ "worker taking from id=" <> show i
+      (t, uNo) <- liftIO $ atomically $ readTBQueue chan
+      traceM $ "worker took from id=" <> show i <> " val=" <> show (t, uNo)
+      sendAndReceive uNo userMap
+      traceM $ "worker end id=" <> show i
+
+  producer <- async $ forM_ fakeData $ \(t, uNo) ->
+    when ((uNo `mod` shardingGroupCount) == shardingGroup)
+      $ let workerShard = fromIntegral uNo `mod` threadCount
+            (i, chan) = tbchans !! workerShard
+         in do
+              waitForTimeStamp t
+              traceM $ "producer putting to shard=" <> show workerShard <> " id=" <> show i <> " val=" <> show (t, uNo)
+              liftIO $ atomically $ writeTBQueue chan (t, uNo)
+
+  liftIO . void . waitAnyCancel $ producer : workers
+
+waitForTimeStamp :: UTCTime -> App ()
+waitForTimeStamp timestamp = liftIO $ do
+  now <- getCurrentTime
+  when (now < timestamp)
+    $
+    -- Event comes from the simulated future: Wait here until now and timestamp are aligned.
+    let delta = diffTimeToMicroSeconds $ diffUTCTime timestamp now
+     in print ("Waiting " ++ show delta ++ " microseconds. (timestamp, now)" ++ show (timestamp, now))
+          >> threadDelay delta
+  where
+    diffTimeToMicroSeconds :: NominalDiffTime -> Int
+    diffTimeToMicroSeconds dt = floor @Double (realToFrac dt * 1_000_000)
+
+plusDelta :: UTCTime -> Word -> UTCTime
+plusDelta timestamp deltaMilliSeconds = addUTCTime (fromIntegral deltaMilliSeconds / 1000) timestamp
+
+sendAndReceive :: Word -> Map Word TestRecipient -> App ()
+sendAndReceive userNo userMap = do
+  print $ "pushing to user" ++ show userNo
+  let testRecipient = userMap Map.! (fromIntegral userNo)
+      alice = testRecipient.user
+
+  r <- recipient alice
+  payload :: Value <- toJSON <$> liftIO randomPayload
+  now <- liftIO $ getCurrentTime
+  let push =
+        object
+          [ "recipients" .= [r],
+            "payload"
+              .= [ object
+                     [ "foo" .= payload,
+                       "sent_at" .= now
+                     ]
+                 ]
+          ]
+
+  void $ postPush alice [push] >>= getBody 200
+
+  traceM $ "pushed to userNo=" <> show userNo <> " push=" <> show push
+
+  messageDeliveryTimeout <- asks $ fromIntegral . (.maxDeliveryDelay)
+  traceM $ "XXX all clientIds " <> show testRecipient.clientIds
+  forM_ (testRecipient.clientIds) $ \(cid :: String) -> do
+    traceM $ "XXX using clientId=" <> show cid
+    runCodensity (TestEvents.createEventsWebSocket alice (Just cid)) $ \ws -> do
+      -- TODO: Tweak this value to the least acceptable event delivery duration
+      local (setTimeoutTo messageDeliveryTimeout) $ TestEvents.assertFindsEvent ws $ \e -> do
+        receivedAt <- liftIO getCurrentTime
+        p <- e %. "data.event.payload.0"
+        sentAt <-
+          (p %. "sent_at" >>= asString)
+            <&> ( \sentAtStr ->
+                    fromMaybe (error ("Cannot parse timestamp: " <> sentAtStr)) $ parseUTC sentAtStr
+                )
+        print $ "Message sent/receive delta: " ++ show (diffUTCTime receivedAt sentAt)
+
+        p %. "foo" `shouldMatch` payload
+      traceM $ "XXX Succeeded for clientId=" <> show cid
+  where
+    -- \| Generate a random string with random length up to 2048 bytes
+    randomPayload :: IO String
+    randomPayload =
+      -- Measured with
+      -- `kubectl exec --namespace databases -it gundeck-gundeck-eks-eu-west-1a-sts-0 -- sh -c 'cqlsh -e "select  blobAsText(payload) from gundeck.notifications LIMIT 5000;" ' | sed 's/^[ \t]*//;s/[ \t]*$//' | wc`
+      let len :: Int = 884 -- measured in prod
+       in mapM (\_ -> randomRIO ('\32', '\126')) [1 .. len] -- printable ASCII
+    parseUTC :: String -> Maybe UTCTime
+    parseUTC = parseTimeM True defaultTimeLocale "%Y-%m-%dT%H:%M:%S%QZ"
+
+setTimeoutTo :: Int -> Env -> Env
+setTimeoutTo tSecs env = env {timeOutSeconds = tSecs}
+
+generateTestRecipient :: (HasCallStack) => App TestRecipient
+generateTestRecipient = do
+  print "generateTestRecipient"
+  user <- recover $ (randomUser OwnDomain def)
+  r <- randomRIO @Int (1, 8)
+  clientIds <- forM [0 .. r] $ \i -> do
+    client <-
+      recover
+        $ addClient
+          user
+          def
+            { acapabilities = Just ["consumable-notifications"],
+              prekeys = Nothing,
+              lastPrekey = Just $ (someLastPrekeysRendered !! i),
+              clabel = "Test Client No. " <> show i,
+              model = "Test Model No. " <> show i
+            }
+        >>= getJSON 201
+    objId client
+
+  pure $ TestRecipient user clientIds
+  where
+    recover :: App a -> App a
+    recover = recoverAll (limitRetriesByCumulativeDelay 300 (exponentialBackoff 1_000_000)) . const

integration/test/Testlib/Env.hs

@@ -62,8 +62,8 @@ serviceHostPort m Stern = m.stern
 serviceHostPort m FederatorInternal = m.federatorInternal
 serviceHostPort m WireServerEnterprise = m.wireServerEnterprise
 
-mkGlobalEnv :: FilePath -> Codensity IO GlobalEnv
-mkGlobalEnv cfgFile = do
+mkGlobalEnv :: FilePath -> Word -> Codensity IO GlobalEnv
+mkGlobalEnv cfgFile shardingGroup = do
   eith <- liftIO $ Yaml.decodeFileEither cfgFile
   intConfig <- liftIO $ case eith of
     Left err -> do
@@ -145,7 +145,11 @@ mkGlobalEnv cfgFile = do
         gDNSMockServerConfig = intConfig.dnsMockServer,
         gCellsEventQueue = intConfig.cellsEventQueue,
         gCellsEventWatchersLock,
-        gCellsEventWatchers
+        gCellsEventWatchers,
+        gShardingGroupCount = intConfig.shardingGroupCount,
+        gShardingGroup = shardingGroup,
+        gMaxUserNo = intConfig.maxUserNo,
+        gMaxDeliveryDelay = intConfig.maxDeliveryDelay
       }
   where
     createSSLContext :: Maybe FilePath -> IO (Maybe OpenSSL.SSLContext)
@@ -201,7 +205,11 @@ mkEnv currentTestName ge = do
           dnsMockServerConfig = ge.gDNSMockServerConfig,
           cellsEventQueue = ge.gCellsEventQueue,
           cellsEventWatchersLock = ge.gCellsEventWatchersLock,
-          cellsEventWatchers = ge.gCellsEventWatchers
+          cellsEventWatchers = ge.gCellsEventWatchers,
+          shardingGroupCount = ge.gShardingGroupCount,
+          shardingGroup = ge.gShardingGroup,
+          maxUserNo = ge.gMaxUserNo,
+          maxDeliveryDelay = ge.gMaxDeliveryDelay
         }
 
 allCiphersuites :: [Ciphersuite]