Introduce NotificationSubsystem (#3786)

This commit introduces the concept of Subsystems. Each of these subsystems will
represent an important part of the domain concepts in the product that will
interact with other subsystems. We will use effect systems to encode these
subsystems and test them in isolation as much as possible.

This commit consolidates all the code that spoke to gundeck from brig and galley
into the NotificationSubsystem.

https://wearezeta.atlassian.net/browse/WPB-5985

---------

Co-authored-by: Magnus Viernickel <magnus.viernickel@wire.com>
Co-authored-by: Leif Battermann <leif.battermann@wire.com>

cabal.project

@@ -31,6 +31,7 @@ packages:
   , libs/wire-api/
   , libs/wire-api-federation/
   , libs/wire-message-proto-lens/
+  , libs/wire-subsystems/
   , libs/zauth/
   , services/background-worker/
   , services/brig/
@@ -162,6 +163,8 @@ package wire-api-federation
     ghc-options: -Werror
 package wire-message-proto-lens
     ghc-options: -Werror
+package wire-subsystems
+    ghc-options: -Werror
 package zauth
     ghc-options: -Werror
 package fedcalls

changelog.d/5-internal/notification-subsystem

@@ -0,0 +1 @@
+Start refactoring code into subsystems, first subsystem being the NotificationSubsystem.

libs/extended/default.nix

@@ -31,6 +31,7 @@
 , servant-server
 , temporary
 , text
+, time
 , tinylog
 , unliftio
 , wai
@@ -63,6 +64,7 @@ mkDerivation {
     servant-openapi3
     servant-server
     text
+    time
     tinylog
     unliftio
     wai

libs/extended/extended.cabal

@@ -19,6 +19,7 @@ build-type:    Simple
 library
   -- cabal-fmt: expand src
   exposed-modules:
+    Data.Time.Clock.DiffTime
     Network.AMQP.Extended
     Network.RabbitMqAdmin
     Options.Applicative.Extended
@@ -101,6 +102,7 @@ library
     , servant-openapi3
     , servant-server
     , text
+    , time
     , tinylog
     , unliftio
     , wai

libs/extended/src/Data/Time/Clock/DiffTime.hs

@@ -0,0 +1,43 @@
+module Data.Time.Clock.DiffTime
+  ( DiffTime,
+    weeksToDiffTime,
+    daysToDiffTime,
+    hoursToDiffTime,
+    minutesToDiffTime,
+    secondsToDiffTime,
+    millisecondsToDiffTime,
+    microsecondsToDiffTime,
+    nanosecondsToDiffTime,
+    picosecondsToDiffTime,
+    diffTimeToFullMicroseconds,
+    diffTimeToPicoseconds,
+  )
+where
+
+import Data.Time
+import Imports
+
+weeksToDiffTime,
+  daysToDiffTime,
+  hoursToDiffTime,
+  minutesToDiffTime,
+  millisecondsToDiffTime,
+  microsecondsToDiffTime,
+  nanosecondsToDiffTime ::
+    Integer -> DiffTime
+weeksToDiffTime = daysToDiffTime . (7 *)
+daysToDiffTime = hoursToDiffTime . (24 *)
+hoursToDiffTime = minutesToDiffTime . (60 *)
+minutesToDiffTime = secondsToDiffTime . (60 *)
+millisecondsToDiffTime = picosecondsToDiffTime . (e9 *)
+microsecondsToDiffTime = picosecondsToDiffTime . (e6 *)
+nanosecondsToDiffTime = picosecondsToDiffTime . (e3 *)
+
+-- | Rounds down. Useful for 'threadDelay', 'timeout', etc.
+diffTimeToFullMicroseconds :: DiffTime -> Int
+diffTimeToFullMicroseconds = fromInteger . (`div` e6) . diffTimeToPicoseconds
+
+e3, e6, e9 :: Integer
+e3 = 1_000
+e6 = 1_000_000
+e9 = 1_000_000_000

libs/gundeck-types/src/Gundeck/Types/Push/V2.hs

@@ -171,14 +171,15 @@ instance ToJSON RecipientClients where
 -- ApsData
 
 newtype ApsSound = ApsSound {fromSound :: Text}
-  deriving (Eq, Show, ToJSON, FromJSON)
+  deriving (Eq, Show, ToJSON, FromJSON, Arbitrary)
 
 newtype ApsLocKey = ApsLocKey {fromLocKey :: Text}
-  deriving (Eq, Show, ToJSON, FromJSON)
+  deriving (Eq, Show, ToJSON, FromJSON, Arbitrary)
 
 data ApsPreference
   = ApsStdPreference
-  deriving (Eq, Show)
+  deriving (Eq, Show, Generic)
+  deriving (Arbitrary) via GenericUniform ApsPreference
 
 instance ToJSON ApsPreference where
   toJSON ApsStdPreference = "std"
@@ -195,7 +196,8 @@ data ApsData = ApsData
     _apsPreference :: !(Maybe ApsPreference),
     _apsBadge :: !Bool
   }
-  deriving (Eq, Show)
+  deriving (Eq, Show, Generic)
+  deriving (Arbitrary) via GenericUniform ApsData
 
 makeLenses ''ApsData
 

libs/polysemy-wire-zoo/polysemy-wire-zoo.cabal

@@ -17,6 +17,7 @@ library
     Wire.Sem.Concurrency
     Wire.Sem.Concurrency.IO
     Wire.Sem.Concurrency.Sequential
+    Wire.Sem.Delay
     Wire.Sem.FromUTC
     Wire.Sem.Jwk
     Wire.Sem.Logger

libs/polysemy-wire-zoo/src/Wire/Sem/Concurrency.hs

@@ -80,6 +80,32 @@ unsafePooledMapConcurrentlyN_ n f as =
     (UnsafePooledMapConcurrentlyN_ n f as :: Concurrency 'Unsafe (Sem r) ())
 {-# INLINEABLE unsafePooledMapConcurrentlyN_ #-}
 
+unsafePooledForConcurrentlyN ::
+  forall r t a b.
+  (Member (Concurrency 'Unsafe) r, Foldable t) =>
+  -- | Max. number of threads. Should not be less than 1.
+  Int ->
+  t a ->
+  (a -> Sem r b) ->
+  Sem r [b]
+unsafePooledForConcurrentlyN n as f =
+  send
+    (UnsafePooledMapConcurrentlyN n f as :: Concurrency 'Unsafe (Sem r) [b])
+{-# INLINEABLE unsafePooledForConcurrentlyN #-}
+
+unsafePooledForConcurrentlyN_ ::
+  forall r t a b.
+  (Member (Concurrency 'Unsafe) r, Foldable t) =>
+  -- | Max. number of threads. Should not be less than 1.
+  Int ->
+  t a ->
+  (a -> Sem r b) ->
+  Sem r ()
+unsafePooledForConcurrentlyN_ n as f =
+  send
+    (UnsafePooledMapConcurrentlyN_ n f as :: Concurrency 'Unsafe (Sem r) ())
+{-# INLINEABLE unsafePooledForConcurrentlyN_ #-}
+
 pooledMapConcurrentlyN ::
   forall r' r t a b.
   r' ~ '[Final IO] =>
@@ -111,3 +137,35 @@ pooledMapConcurrentlyN_ n f as =
         Concurrency 'Safe (Sem r) ()
     )
 {-# INLINEABLE pooledMapConcurrentlyN_ #-}
+
+pooledForConcurrentlyN ::
+  forall r' r t a b.
+  r' ~ '[Final IO] =>
+  (Member (Concurrency 'Safe) r, Subsume r' r, Foldable t) =>
+  -- | Max. number of threads. Should not be less than 1.
+  Int ->
+  t a ->
+  (a -> Sem r' b) ->
+  Sem r [b]
+pooledForConcurrentlyN n as f =
+  send
+    ( UnsafePooledMapConcurrentlyN n (subsume_ @r' @r . f) as ::
+        Concurrency 'Safe (Sem r) [b]
+    )
+{-# INLINEABLE pooledForConcurrentlyN #-}
+
+pooledForConcurrentlyN_ ::
+  forall r' r t a b.
+  r' ~ '[Final IO] =>
+  (Member (Concurrency 'Safe) r, Subsume r' r, Foldable t) =>
+  -- | Max. number of threads. Should not be less than 1.
+  Int ->
+  t a ->
+  (a -> Sem r' b) ->
+  Sem r ()
+pooledForConcurrentlyN_ n as f =
+  send
+    ( UnsafePooledMapConcurrentlyN_ n (subsume_ @r' @r . f) as ::
+        Concurrency 'Safe (Sem r) ()
+    )
+{-# INLINEABLE pooledForConcurrentlyN_ #-}

libs/polysemy-wire-zoo/src/Wire/Sem/Delay.hs

@@ -0,0 +1,32 @@
+{-# LANGUAGE TemplateHaskell #-}
+
+module Wire.Sem.Delay where
+
+import Imports
+import Polysemy
+
+data Delay m a where
+  Delay :: Int -> Delay m ()
+
+makeSem ''Delay
+
+runDelay :: Member (Embed IO) r => Sem (Delay ': r) a -> Sem r a
+runDelay = interpret $ \case
+  Delay i -> threadDelay i
+
+runControlledDelay :: forall r a. (Member (Embed IO) r) => MVar Int -> Sem (Delay : r) a -> Sem r a
+runControlledDelay tickSource = interpret $ \case
+  Delay n -> waitForTicks n
+  where
+    waitForTicks :: Int -> Sem r ()
+    waitForTicks 0 = pure ()
+    waitForTicks remaining0 = do
+      passedTicks <- takeMVar tickSource
+      let remaining = remaining0 - passedTicks
+      if remaining <= 0
+        then pure ()
+        else waitForTicks remaining
+
+runDelayInstantly :: Sem (Delay : r) a -> Sem r a
+runDelayInstantly = interpret $ \case
+  Delay _ -> pure ()

libs/polysemy-wire-zoo/src/Wire/Sem/Logger/TinyLog.hs

@@ -1,3 +1,5 @@
+{-# LANGUAGE RecordWildCards #-}
+
 -- This file is part of the Wire Server implementation.
 --
 -- Copyright (C) 2022 Wire Swiss GmbH <opensource@wire.com>
@@ -21,12 +23,16 @@ module Wire.Sem.Logger.TinyLog
     stringLoggerToTinyLog,
     discardTinyLogs,
     module Wire.Sem.Logger.Level,
+    LogRecorder (..),
+    newLogRecorder,
+    recordLogs,
   )
 where
 
 import Data.Id
 import Imports
 import Polysemy
+import Polysemy.TinyLog (TinyLog)
 import qualified System.Logger as Log
 import Wire.Sem.Logger
 import Wire.Sem.Logger.Level
@@ -58,3 +64,12 @@ stringLoggerToTinyLog = mapLogger @String Log.msg
 
 discardTinyLogs :: Sem (Logger (Log.Msg -> Log.Msg) ': r) a -> Sem r a
 discardTinyLogs = discardLogs
+
+newtype LogRecorder = LogRecorder {recordedLogs :: IORef [(Level, LByteString)]}
+
+newLogRecorder :: IO LogRecorder
+newLogRecorder = LogRecorder <$> newIORef []
+
+recordLogs :: Member (Embed IO) r => LogRecorder -> Sem (TinyLog ': r) a -> Sem r a
+recordLogs LogRecorder {..} = interpret $ \(Log lvl msg) ->
+  modifyIORef' recordedLogs (++ [(lvl, Log.render (Log.renderDefault ", ") msg)])

libs/types-common/src/Data/Range.hs

@@ -508,6 +508,9 @@ genRange pack_ gc =
 instance (KnownNat n, KnownNat m, n <= m) => Arbitrary (Range n m Integer) where
   arbitrary = genIntegral
 
+instance (KnownNat n, KnownNat m, n <= m) => Arbitrary (Range n m Int32) where
+  arbitrary = genIntegral
+
 instance (KnownNat n, KnownNat m, n <= m) => Arbitrary (Range n m Word) where
   arbitrary = genIntegral