brig: Use NotificationSubsystem

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_ #-}

nix/haskell-pins.nix

@@ -283,10 +283,6 @@ let
       version = "0.4.0";
       sha256 = "sha256-DSMckKIeVE/buSMg8Mq+mUm1bYPYB7veA11Ns7vTBbc=";
     };
-    polysemy = {
-      version = "1.8.0.0";
-      sha256 = "sha256-AdxxKWXdUjZiHLDj6iswMWpycs7mFB8eKhBR4ljF6kk=";
-    };
     hpack = {
       version = "0.36.0";
       sha256 = "sha256-a8jKkzO3CWIoBg+Uaw5TtpDwmeajWCTW1zJNrlpBKPU=";

nix/manual-overrides.nix

@@ -22,9 +22,11 @@ hself: hsuper: {
   network-arbitrary = hlib.markUnbroken (hlib.doJailbreak hsuper.network-arbitrary);
   one-liner = hlib.doJailbreak hsuper.one-liner;
   linear-generics = hsuper.linear-generics_0_2_2;
-  polysemy = hlib.doJailbreak hsuper.polysemy;
+  # polysemy = hlib.doJailbreak hsuper.polysemy;
+  polysemy-test = hlib.markUnbroken hsuper.polysemy-test;
+  # polysemy-resume = hlib.doJailbreak hsuper.polysemy-resume;
   polysemy-check = hlib.markUnbroken (hlib.doJailbreak hsuper.polysemy-check);
-  polysemy-plugin = hlib.doJailbreak hsuper.polysemy-plugin;
+  # polysemy-plugin = hlib.doJailbreak hsuper.polysemy-plugin;
   quickcheck-state-machine = hlib.dontCheck hsuper.quickcheck-state-machine;
   servant = hlib.doJailbreak hsuper.servant;
   servant-client = hlib.doJailbreak hsuper.servant-client;

services/brig/brig.cabal

@@ -313,7 +313,9 @@ library
     , openapi3
     , optparse-applicative       >=0.11
     , polysemy
+    , polysemy-conc
     , polysemy-plugin
+    , polysemy-time
     , polysemy-wire-zoo
     , proto-lens                 >=0.1
     , random-shuffle             >=0.0.3
@@ -361,6 +363,7 @@ library
     , wai-utilities              >=0.16
     , wire-api
     , wire-api-federation
+    , wire-subsystems
     , yaml                       >=0.8.22
     , zauth                      >=0.10.3
 

services/brig/default.nix

@@ -89,7 +89,9 @@
 , pem
 , pipes
 , polysemy
+, polysemy-conc
 , polysemy-plugin
+, polysemy-time
 , polysemy-wire-zoo
 , postie
 , process
@@ -155,6 +157,7 @@
 , warp-tls
 , wire-api
 , wire-api-federation
+, wire-subsystems
 , yaml
 , zauth
 }:
@@ -236,7 +239,9 @@ mkDerivation {
     openapi3
     optparse-applicative
     polysemy
+    polysemy-conc
     polysemy-plugin
+    polysemy-time
     polysemy-wire-zoo
     proto-lens
     random-shuffle
@@ -284,6 +289,7 @@ mkDerivation {
     wai-utilities
     wire-api
     wire-api-federation
+    wire-subsystems
     yaml
     zauth
   ];

services/brig/src/Brig/API/Auth.hs

@@ -43,13 +43,21 @@ import Network.HTTP.Types
 import Network.Wai.Utilities ((!>>))
 import Network.Wai.Utilities.Error qualified as Wai
 import Polysemy
+import Polysemy.Async
+import Polysemy.TinyLog (TinyLog)
 import Wire.API.User
 import Wire.API.User.Auth hiding (access)
 import Wire.API.User.Auth.LegalHold
 import Wire.API.User.Auth.ReAuth
 import Wire.API.User.Auth.Sso
+import Wire.NotificationSubsystem
 
 accessH ::
+  ( Member TinyLog r,
+    Member (Embed HttpClientIO) r,
+    Member NotificationSubsystem r,
+    Member Async r
+  ) =>
   Maybe ClientId ->
   [Either Text SomeUserToken] ->
   Maybe (Either Text SomeAccessToken) ->
@@ -61,22 +69,36 @@ accessH mcid ut' mat' = do
     >>= either (uncurry (access mcid)) (uncurry (access mcid))
 
 access ::
-  (TokenPair u a) =>
+  ( TokenPair u a,
+    Member TinyLog r,
+    Member (Embed HttpClientIO) r,
+    Member NotificationSubsystem r,
+    Member Async r
+  ) =>
   Maybe ClientId ->
   NonEmpty (Token u) ->
   Maybe (Token a) ->
   Handler r SomeAccess
 access mcid t mt =
   traverse mkUserTokenCookie
-    =<< wrapHttpClientE (Auth.renewAccess (List1 t) mt mcid) !>> zauthError
+    =<< Auth.renewAccess (List1 t) mt mcid !>> zauthError
 
-sendLoginCode :: SendLoginCode -> Handler r LoginCodeTimeout
+sendLoginCode :: (Member TinyLog r) => SendLoginCode -> Handler r LoginCodeTimeout
 sendLoginCode (SendLoginCode phone call force) = do
   checkAllowlist (Right phone)
-  c <- wrapClientE (Auth.sendLoginCode phone call force) !>> sendLoginCodeError
+  c <- Auth.sendLoginCode phone call force !>> sendLoginCodeError
   pure $ LoginCodeTimeout (pendingLoginTimeout c)
 
-login :: (Member GalleyProvider r) => Login -> Maybe Bool -> Handler r SomeAccess
+login ::
+  ( Member GalleyProvider r,
+    Member TinyLog r,
+    Member (Embed HttpClientIO) r,
+    Member NotificationSubsystem r,
+    Member Async r
+  ) =>
+  Login ->
+  Maybe Bool ->
+  Handler r SomeAccess
 login l (fromMaybe False -> persist) = do
   let typ = if persist then PersistentCookie else SessionCookie
   c <- Auth.login l typ !>> loginError
@@ -94,7 +116,7 @@ logoutH uts' mat' = do
 
 logout :: TokenPair u a => NonEmpty (Token u) -> Maybe (Token a) -> Handler r ()
 logout _ Nothing = throwStd authMissingToken
-logout uts (Just at) = wrapHttpClientE $ Auth.logout (List1 uts) at !>> zauthError
+logout uts (Just at) = Auth.logout (List1 uts) at !>> zauthError
 
 changeSelfEmailH ::
   Member BlacklistStore r =>
@@ -117,32 +139,48 @@ validateCredentials ::
   Handler r UserId
 validateCredentials _ Nothing = throwStd missingAccessToken
 validateCredentials uts mat =
-  fst <$> wrapHttpClientE (Auth.validateTokens (List1 uts) mat) !>> zauthError
+  fst <$> Auth.validateTokens (List1 uts) mat !>> zauthError
 
 listCookies :: Local UserId -> Maybe (CommaSeparatedList CookieLabel) -> Handler r CookieList
 listCookies lusr (fold -> labels) =
   CookieList
     <$> wrapClientE (Auth.listCookies (tUnqualified lusr) (toList labels))
 
-removeCookies :: Local UserId -> RemoveCookies -> Handler r ()
+removeCookies :: (Member TinyLog r) => Local UserId -> RemoveCookies -> Handler r ()
 removeCookies lusr (RemoveCookies pw lls ids) =
-  wrapClientE (Auth.revokeAccess (tUnqualified lusr) pw ids lls) !>> authError
+  Auth.revokeAccess (tUnqualified lusr) pw ids lls !>> authError
 
-legalHoldLogin :: (Member GalleyProvider r) => LegalHoldLogin -> Handler r SomeAccess
+legalHoldLogin ::
+  ( Member GalleyProvider r,
+    Member (Embed HttpClientIO) r,
+    Member NotificationSubsystem r,
+    Member Async r,
+    Member TinyLog r
+  ) =>
+  LegalHoldLogin ->
+  Handler r SomeAccess
 legalHoldLogin lhl = do
   let typ = PersistentCookie -- Session cookie isn't a supported use case here
   c <- Auth.legalHoldLogin lhl typ !>> legalHoldLoginError
   traverse mkUserTokenCookie c
 
-ssoLogin :: SsoLogin -> Maybe Bool -> Handler r SomeAccess
+ssoLogin ::
+  ( Member TinyLog r,
+    Member (Embed HttpClientIO) r,
+    Member NotificationSubsystem r,
+    Member Async r
+  ) =>
+  SsoLogin ->
+  Maybe Bool ->
+  Handler r SomeAccess
 ssoLogin l (fromMaybe False -> persist) = do
   let typ = if persist then PersistentCookie else SessionCookie
-  c <- wrapHttpClientE (Auth.ssoLogin l typ) !>> loginError
+  c <- Auth.ssoLogin l typ !>> loginError
   traverse mkUserTokenCookie c
 
-getLoginCode :: Phone -> Handler r PendingLoginCode
+getLoginCode :: (Member TinyLog r) => Phone -> Handler r PendingLoginCode
 getLoginCode phone = do
-  code <- lift $ wrapClient $ Auth.lookupLoginCode phone
+  code <- lift $ Auth.lookupLoginCode phone
   maybe (throwStd loginCodeNotFound) pure code
 
 reauthenticate :: Member GalleyProvider r => UserId -> ReAuthUser -> Handler r ()

services/brig/src/Brig/API/Client.hs

@@ -88,7 +88,8 @@ import Data.Set qualified as Set
 import Imports
 import Network.HTTP.Types.Method (StdMethod)
 import Network.Wai.Utilities
-import Polysemy (Member)
+import Polysemy
+import Polysemy.Async
 import Servant (Link, ToHttpApiData (toUrlPiece))
 import System.Logger.Class (field, msg, val, (~~))
 import System.Logger.Class qualified as Log
@@ -104,6 +105,7 @@ import Wire.API.User.Client
 import Wire.API.User.Client.DPoPAccessToken
 import Wire.API.User.Client.Prekey
 import Wire.API.UserMap (QualifiedUserMap (QualifiedUserMap, qualifiedUserMap), UserMap (userMap))
+import Wire.NotificationSubsystem
 import Wire.Sem.Concurrency
 import Wire.Sem.FromUTC (FromUTC (fromUTCTime))
 import Wire.Sem.Now as Now
@@ -152,7 +154,11 @@ lookupLocalPubClientsBulk :: [UserId] -> ExceptT ClientError (AppT r) (UserMap (
 lookupLocalPubClientsBulk = lift . wrapClient . Data.lookupPubClientsBulk
 
 addClient ::
-  (Member GalleyProvider r) =>
+  ( Member GalleyProvider r,
+    Member (Embed HttpClientIO) r,
+    Member NotificationSubsystem r,
+    Member Async r
+  ) =>
   UserId ->
   Maybe ConnId ->
   NewClient ->
@@ -163,7 +169,11 @@ addClient = addClientWithReAuthPolicy Data.reAuthForNewClients
 -- a superset of the clients known to galley.
 addClientWithReAuthPolicy ::
   forall r.
-  (Member GalleyProvider r) =>
+  ( Member GalleyProvider r,
+    Member (Embed HttpClientIO) r,
+    Member NotificationSubsystem r,
+    Member Async r
+  ) =>
   Data.ReAuthPolicy ->
   UserId ->
   Maybe ConnId ->
@@ -190,8 +200,8 @@ addClientWithReAuthPolicy policy u con new = do
   lift $ do
     for_ old $ execDelete u con
     liftSem $ GalleyProvider.newClient u (clientId clt)
-    wrapHttp $ Intra.onClientEvent u con (ClientAdded u clt)
-    when (clientType clt == LegalHoldClientType) $ wrapHttpClient $ Intra.onUserEvent u con (UserLegalHoldEnabled u)
+    liftSem $ Intra.onClientEvent u con (ClientAdded u clt)
+    when (clientType clt == LegalHoldClientType) $ liftSem $ Intra.onUserEvent u con (UserLegalHoldEnabled u)
     when (count > 1) $
       for_ (userEmail usr) $
         \email ->
@@ -461,9 +471,16 @@ pubClient c =
       pubClientClass = clientClass c
     }
 
-legalHoldClientRequested :: UserId -> LegalHoldClientRequest -> (AppT r) ()
+legalHoldClientRequested ::
+  ( Member (Embed HttpClientIO) r,
+    Member NotificationSubsystem r,
+    Member Async r
+  ) =>
+  UserId ->
+  LegalHoldClientRequest ->
+  AppT r ()
 legalHoldClientRequested targetUser (LegalHoldClientRequest _requester lastPrekey') =
-  wrapHttpClient $ Intra.onUserEvent targetUser Nothing lhClientEvent
+  liftSem $ Intra.onUserEvent targetUser Nothing lhClientEvent
   where
     clientId :: ClientId
     clientId = clientIdFromPrekey $ unpackLastPrekey lastPrekey'
@@ -472,14 +489,20 @@ legalHoldClientRequested targetUser (LegalHoldClientRequest _requester lastPreke
     lhClientEvent :: UserEvent
     lhClientEvent = LegalHoldClientRequested eventData
 
-removeLegalHoldClient :: UserId -> (AppT r) ()
+removeLegalHoldClient ::
+  ( Member (Embed HttpClientIO) r,
+    Member NotificationSubsystem r,
+    Member Async r
+  ) =>
+  UserId ->
+  AppT r ()
 removeLegalHoldClient uid = do
   clients <- wrapClient $ Data.lookupClients uid
   -- Should only be one; but just in case we'll treat it as a list
   let legalHoldClients = filter ((== LegalHoldClientType) . clientType) clients
   -- maybe log if this isn't the case
   forM_ legalHoldClients (execDelete uid Nothing)
-  wrapHttpClient $ Intra.onUserEvent uid Nothing (UserLegalHoldDisabled uid)
+  liftSem $ Intra.onUserEvent uid Nothing (UserLegalHoldDisabled uid)
 
 createAccessToken ::
   (Member JwtTools r, Member Now r, Member PublicKeyBundle r) =>