Introduce NotificationSubsystem to push notifications

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