simplexmq/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs

{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedLists #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TupleSections #-}
{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}

module Simplex.Messaging.Agent.NtfSubSupervisor
  ( runNtfSupervisor,
    nsUpdateToken,
    nsRemoveNtfToken,
    sendNtfSubCommand,
    hasInstantNotifications,
    instantNotifications,
    deleteToken,
    closeNtfSupervisor,
    getNtfServer,
  )
where

import Control.Logger.Simple (logError, logInfo)
import Control.Monad
import Control.Monad.Reader
import Control.Monad.Trans.Except
import Crypto.Random (ChaChaDRG)
import Data.Bifunctor (first)
import Data.Either (fromRight, partitionEithers)
import Data.Functor (($>))
import Data.List (foldl')
import Data.List.NonEmpty (NonEmpty (..))
import qualified Data.List.NonEmpty as L
import qualified Data.Map.Strict as M
import Data.Maybe (catMaybes)
import qualified Data.Set as S
import qualified Data.Text as T
import Data.Time (UTCTime, addUTCTime, getCurrentTime)
import Data.Time.Clock (diffUTCTime)
import Simplex.Messaging.Agent.Client
import Simplex.Messaging.Agent.Env.SQLite
import Simplex.Messaging.Agent.Protocol
import Simplex.Messaging.Agent.RetryInterval
import Simplex.Messaging.Agent.Stats
import Simplex.Messaging.Agent.Store
import Simplex.Messaging.Agent.Store.AgentStore
import qualified Simplex.Messaging.Agent.Store.DB as DB
import Simplex.Messaging.Client (NetworkRequestMode (..), nonBlockingWriteTBQueue)
import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Notifications.Protocol
import Simplex.Messaging.Notifications.Types
import Simplex.Messaging.Protocol (NtfServer, sameSrvAddr)
import qualified Simplex.Messaging.Protocol as SMP
import Simplex.Messaging.Util (catchAllErrors, catchAllErrors', diffToMicroseconds, threadDelay', tryAllErrors, tshow, whenM)
import System.Random (randomR)
import UnliftIO
import UnliftIO.Concurrent (forkIO)
import qualified UnliftIO.Exception as E

runNtfSupervisor :: AgentClient -> AM' ()
runNtfSupervisor c = do
  ns <- asks ntfSupervisor
  runExceptT startTknDelete >>= \case
    Left e -> notifyErr e
    Right _ -> pure ()
  forever $ do
    cmd <- atomically . readTBQueue $ ntfSubQ ns
    handleErr $ agentOperationBracket c AONtfNetwork waitUntilActive $
      processNtfCmd c cmd `catchAllErrors'` notifyErr
  where
    startTknDelete :: AM ()
    startTknDelete = do
      pendingDelServers <- withStore' c getPendingDelTknServers
      lift . forM_ pendingDelServers $ getNtfTknDelWorker True c
    handleErr :: AM' () -> AM' ()
    handleErr = E.handle $ \(e :: E.SomeException) -> notifyErr e
    notifyErr e = notifyInternalError' c $ "runNtfSupervisor error " <> show e

partitionErrs :: (a -> ConnId) -> [a] -> [Either AgentErrorType b] -> ([(ConnId, AgentErrorType)], [b])
partitionErrs f xs = partitionEithers . zipWith (\x -> first (f x,)) xs
{-# INLINE partitionErrs #-}

ntfSubConnId :: NtfSubscription -> ConnId
ntfSubConnId NtfSubscription {connId} = connId

processNtfCmd :: AgentClient -> (NtfSupervisorCommand, NonEmpty ConnId) -> AM ()
processNtfCmd c (cmd, connIds) = do
  logInfo $ "processNtfCmd - cmd = " <> tshow cmd
  let connIds' = L.toList connIds
  case cmd of
    NSCCreate -> do
      (cErrs, rqSubActions) <- lift $ partitionErrs id connIds' <$> withStoreBatch c (\db -> map (getQueueSub db) connIds')
      notifyErrs c cErrs
      logInfo $ "processNtfCmd, NSCCreate - length rqSubs = " <> tshow (length rqSubActions)
      let (ns, rs, css, cns) = partitionQueueSubActions rqSubActions
      createNewSubs ns
      resetSubs rs
      lift $ do
        mapM_ (getNtfSMPWorker True c) (S.fromList css)
        mapM_ (getNtfNTFWorker True c) (S.fromList cns)
      where
        getQueueSub ::
          DB.Connection ->
          ConnId ->
          IO (Either AgentErrorType (RcvQueue, Maybe NtfSupervisorSub))
        getQueueSub db connId = fmap (first storeError) $ runExceptT $ do
          rq <- ExceptT $ getPrimaryRcvQueue db connId
          sub <- liftIO $ getNtfSubscription db connId
          pure (rq, sub)
        createNewSubs :: [RcvQueue] -> AM ()
        createNewSubs rqs = do
          withTokenServer $ \ntfServer -> do
            let newSubs = map (rqToNewSub ntfServer) rqs
            (cErrs, _) <- lift $ partitionErrs ntfSubConnId newSubs <$> withStoreBatch c (\db -> map (storeNewSub db) newSubs)
            notifyErrs c cErrs
            kickSMPWorkers rqs
          where
            rqToNewSub :: NtfServer -> RcvQueue -> NtfSubscription
            rqToNewSub ntfServer RcvQueue {userId, connId, server} = newNtfSubscription userId connId server Nothing ntfServer NASNew
            storeNewSub :: DB.Connection -> NtfSubscription -> IO (Either AgentErrorType ())
            storeNewSub db sub = first storeError <$> createNtfSubscription db sub (NSASMP NSASmpKey)
        resetSubs :: [(RcvQueue, NtfSubscription)] -> AM ()
        resetSubs rqSubs = do
          withTokenServer $ \ntfServer -> do
            let subsToReset = map (toResetSub ntfServer) rqSubs
            (cErrs, _) <- lift $ partitionErrs ntfSubConnId subsToReset <$> withStoreBatch' c (\db -> map (storeResetSub db) subsToReset)
            notifyErrs c cErrs
            let rqs = map fst rqSubs
            kickSMPWorkers rqs
          where
            toResetSub :: NtfServer -> (RcvQueue, NtfSubscription) -> NtfSubscription
            toResetSub ntfServer (rq, sub) =
              let RcvQueue {server = smpServer} = rq
               in sub {smpServer, ntfQueueId = Nothing, ntfServer, ntfSubId = Nothing, ntfSubStatus = NASNew}
            storeResetSub :: DB.Connection -> NtfSubscription -> IO ()
            storeResetSub db sub = supervisorUpdateNtfSub db sub (NSASMP NSASmpKey)
        partitionQueueSubActions ::
          [(RcvQueue, Maybe NtfSupervisorSub)] ->
          ( [RcvQueue], -- new subs
            [(RcvQueue, NtfSubscription)], -- reset subs
            [SMPServer], -- continue work (SMP)
            [NtfServer] -- continue work (Ntf)
          )
        partitionQueueSubActions = foldr decideSubWork ([], [], [], [])
          where
            -- sub = Nothing, needs to be created
            decideSubWork (rq, Nothing) (ns, rs, css, cns) = (rq : ns, rs, css, cns)
            decideSubWork (rq, Just (sub, subAction_)) (ns, rs, css, cns) =
              case (clientNtfCreds rq, ntfQueueId sub) of
                -- notifier ID created on SMP server (on ntf server subscription can be registered or not yet),
                -- need to clarify action
                (Just ClientNtfCreds {notifierId}, Just ntfQueueId')
                  | sameSrvAddr (qServer rq) subSMPServer && notifierId == ntfQueueId' -> contOrReset
                  | otherwise -> reset
                (Nothing, Nothing) -> contOrReset
                _ -> reset
              where
                NtfSubscription {ntfServer = subNtfServer, smpServer = subSMPServer} = sub
                contOrReset = case subAction_ of
                  -- action was set to NULL after worker internal error
                  Nothing -> reset
                  Just (action, _)
                    -- subscription was marked for deletion / is being deleted
                    | isDeleteNtfSubAction action -> reset
                    -- continue work on subscription (e.g. supervisor was repeatedly tasked with creating a subscription)
                    | otherwise -> case action of
                        NSASMP _ -> (ns, rs, qServer rq : css, cns)
                        NSANtf _ -> (ns, rs, css, subNtfServer : cns)
                reset = (ns, (rq, sub) : rs, css, cns)
    NSCSmpDelete -> do
      (cErrs, rqs) <- lift $ partitionErrs id connIds' <$> withStoreBatch c (\db -> map (getQueue db) connIds')
      logInfo $ "processNtfCmd, NSCSmpDelete - length rqs = " <> tshow (length rqs)
      (cErrs', _) <- lift $ partitionErrs qConnId rqs <$> withStoreBatch' c (\db -> map (updateAction db) rqs)
      notifyErrs c (cErrs <> cErrs')
      kickSMPWorkers rqs
      where
        getQueue :: DB.Connection -> ConnId -> IO (Either AgentErrorType RcvQueue)
        getQueue db connId = first storeError <$> getPrimaryRcvQueue db connId
        updateAction :: DB.Connection -> RcvQueue -> IO ()
        updateAction db rq = supervisorUpdateNtfAction db (qConnId rq) (NSASMP NSASmpDelete)
    NSCDeleteSub -> void $ lift $ withStoreBatch' c $ \db -> map (deleteNtfSubscription' db) connIds'
  where
    kickSMPWorkers :: [RcvQueue] -> AM ()
    kickSMPWorkers rqs = do
      let smpServers = S.fromList $ map qServer rqs
      lift $ mapM_ (getNtfSMPWorker True c) smpServers

getNtfNTFWorker :: Bool -> AgentClient -> NtfServer -> AM' Worker
getNtfNTFWorker hasWork c server = do
  ws <- asks $ ntfWorkers . ntfSupervisor
  getAgentWorker "ntf_ntf" hasWork c server ws $ runNtfWorker c server

getNtfSMPWorker :: Bool -> AgentClient -> SMPServer -> AM' Worker
getNtfSMPWorker hasWork c server = do
  ws <- asks $ ntfSMPWorkers . ntfSupervisor
  getAgentWorker "ntf_smp" hasWork c server ws $ runNtfSMPWorker c server

getNtfTknDelWorker :: Bool -> AgentClient -> NtfServer -> AM' Worker
getNtfTknDelWorker hasWork c server = do
  ws <- asks $ ntfTknDelWorkers . ntfSupervisor
  getAgentWorker "ntf_tkn_del" hasWork c server ws $ runNtfTknDelWorker c server

withTokenServer :: (NtfServer -> AM ()) -> AM ()
withTokenServer action = lift getNtfToken >>= mapM_ (\NtfToken {ntfServer} -> action ntfServer)

runNtfWorker :: AgentClient -> NtfServer -> Worker -> AM ()
runNtfWorker c srv Worker {doWork} =
  forever $ do
    waitForWork doWork
    ExceptT $ agentOperationBracket c AONtfNetwork throwWhenInactive $ runExceptT runNtfOperation
  where
    runNtfOperation :: AM ()
    runNtfOperation = do
      ntfBatchSize <- asks $ ntfBatchSize . config
      withWorkItems c doWork (withStore' c $ \db -> getNextNtfSubNTFActions db srv ntfBatchSize) $ \nextSubs -> do
        logInfo $ "runNtfWorker - length nextSubs = " <> tshow (length nextSubs)
        currTs <- liftIO getCurrentTime
        let (creates, checks, deletes, rotates) = splitActions currTs nextSubs
        if null creates && null checks && null deletes && null rotates
          then
            let (_, _, firstActionTs) = L.head nextSubs
             in lift $ rescheduleWork doWork currTs firstActionTs
          else do
            retrySubActions c creates createSubs
            retrySubActions c checks checkSubs
            retrySubActions c deletes deleteSubs
            retrySubActions c rotates rotateSubs
    splitActions :: UTCTime -> NonEmpty (NtfSubNTFAction, NtfSubscription, NtfActionTs) -> ([NtfSubscription], [NtfSubscription], [NtfSubscription], [NtfSubscription])
    splitActions currTs = foldr addAction ([], [], [], [])
      where
        addAction (cmd, sub, ts) acc@(creates, checks, deletes, rotates) = case cmd of
          NSACreate -> (sub : creates, checks, deletes, rotates)
          NSACheck
            | ts <= currTs -> (creates, sub : checks, deletes, rotates)
            | otherwise -> acc
          NSADelete -> (creates, checks, sub : deletes, rotates)
          NSARotate -> (creates, checks, deletes, sub : rotates)
    createSubs :: [NtfSubscription] -> AM' [NtfSubscription]
    createSubs ntfSubs =
      getNtfToken >>= \case
        Just tkn@NtfToken {ntfServer, ntfTokenId = Just tknId, ntfTknStatus = NTActive, ntfMode = NMInstant} -> do
          subsRqs_ <- zip ntfSubs <$> withStoreBatch c (\db -> map (getQueue db) ntfSubs)
          let (errs1, subs_, newSubs_) = splitSubs tknId subsRqs_
          incStatByUserId ntfServer ntfCreateAttempts subs_
          case (L.nonEmpty subs_, L.nonEmpty newSubs_) of
            (Just subs, Just newSubs) -> do
              rs <- L.zip subs <$> agentNtfCreateSubscriptions c tkn newSubs
              let (ntfSubs', errs2, nSubIds) = splitResults $ L.toList rs
                  subs' = map fst nSubIds
                  errs2' = map (first ntfSubConnId) errs2
              incStatByUserId ntfServer ntfCreated subs'
              ts <- liftIO getCurrentTime
              int <- asks $ ntfSubFirstCheckInterval . config
              let checkTs = addUTCTime int ts
              (errs3, _) <- partitionErrs ntfSubConnId subs' <$> withStoreBatch' c (\db -> map (updateSubNSACheck db checkTs) nSubIds)
              workerErrors c $ errs1 <> errs2' <> errs3
              pure ntfSubs'
            _ -> workerErrors c errs1 $> []
        _ -> do
          let errs = map (\sub -> (ntfSubConnId sub, INTERNAL "NSACreate - no active token")) ntfSubs
          workerErrors c errs
          pure []
      where
        getQueue :: DB.Connection -> NtfSubscription -> IO (Either AgentErrorType RcvQueue)
        getQueue db NtfSubscription {connId} = first storeError <$> getPrimaryRcvQueue db connId
        splitSubs :: NtfTokenId -> [(NtfSubscription, Either AgentErrorType RcvQueue)] -> ([(ConnId, AgentErrorType)], [NtfSubscription], [NewNtfEntity 'Subscription])
        splitSubs tknId = foldr splitSub ([], [], [])
          where
            splitSub (sub, rq) (errs, subs, newSubs) = case rq of
              Right RcvQueue {clientNtfCreds = Just creds} -> (errs, sub : subs, toNewSub sub creds : newSubs)
              Right _ -> ((ntfSubConnId sub, INTERNAL "NSACreate - no notifier queue credentials") : errs, subs, newSubs)
              Left e -> ((ntfSubConnId sub, e) : errs, subs, newSubs)
            toNewSub NtfSubscription {smpServer} ClientNtfCreds {ntfPrivateKey, notifierId} =
              NewNtfSub tknId (SMPQueueNtf smpServer notifierId) ntfPrivateKey
        updateSubNSACheck :: DB.Connection -> UTCTime -> (NtfSubscription, NtfSubscriptionId) -> IO ()
        updateSubNSACheck db checkTs (sub, nSubId) = updateNtfSubscription db sub {ntfSubId = Just nSubId, ntfSubStatus = NASCreated NSNew} (NSANtf NSACheck) checkTs
    checkSubs :: [NtfSubscription] -> AM' [NtfSubscription]
    checkSubs ntfSubs =
      getNtfToken >>= \case
        Just tkn@NtfToken {ntfServer, ntfTknStatus = NTActive, ntfMode = NMInstant} -> do
          let (errs1, subs_, subIds_) = splitSubs ntfSubs
          incStatByUserId ntfServer ntfCheckAttempts subs_
          case (L.nonEmpty subs_, L.nonEmpty subIds_) of
            (Just subs, Just subIds) -> do
              rs <- L.zip subs <$> agentNtfCheckSubscriptions c tkn subIds
              let (ntfSubs', errs2, nSubStatuses) = splitResults $ L.toList rs
                  subs' = map fst nSubStatuses
                  (errs2', authSubs) = partitionEithers $ map (\case (sub, NTF _ SMP.AUTH) -> Right sub; e -> Left $ first ntfSubConnId e) errs2
              incStatByUserId ntfServer ntfChecked subs'
              ts <- liftIO getCurrentTime
              int <- asks $ ntfSubCheckInterval . config
              let nextCheckTs = addUTCTime int ts
              (errs3, srvs) <- partitionErrs ntfSubConnId subs' <$> withStoreBatch' c (\db -> map (updateSub db ntfServer ts nextCheckTs) nSubStatuses)
              (errs4, srvs') <- partitionErrs ntfSubConnId authSubs <$> withStoreBatch' c (\db -> map (recreateNtfSub db ntfServer ts) authSubs)
              mapM_ (getNtfSMPWorker True c) $ S.fromList (catMaybes srvs <> srvs')
              workerErrors c $ errs1 <> errs2' <> errs3 <> errs4
              pure ntfSubs'
            _ -> workerErrors c errs1 $> []
        _ -> do
          let errs = map (\sub -> (ntfSubConnId sub, INTERNAL "NSACheck - no active token")) ntfSubs
          workerErrors c errs
          pure []
      where
        splitSubs :: [NtfSubscription] -> ([(ConnId, AgentErrorType)], [NtfSubscription], [NtfSubscriptionId])
        splitSubs = foldr splitSub ([], [], [])
          where
            splitSub sub (errs, subs, subIds) = case sub of
              NtfSubscription {ntfSubId = Just subId} -> (errs, sub : subs, subId : subIds)
              _ -> ((ntfSubConnId sub, INTERNAL "NSACheck - no subscription ID") : errs, subs, subIds)
        updateSub :: DB.Connection -> NtfServer -> UTCTime -> UTCTime -> (NtfSubscription, NtfSubStatus) -> IO (Maybe SMPServer)
        updateSub db ntfServer ts nextCheckTs (sub, status)
          | status `elem` subscribeNtfStatuses =
              let sub' = sub {ntfSubStatus = NASCreated status}
               in Nothing <$ updateNtfSubscription db sub' (NSANtf NSACheck) nextCheckTs
          -- ntf server stopped subscribing to this queue
          | otherwise = Just <$> recreateNtfSub db ntfServer ts sub
        recreateNtfSub :: DB.Connection -> NtfServer -> UTCTime -> NtfSubscription -> IO SMPServer
        recreateNtfSub db ntfServer ts sub@NtfSubscription {smpServer} =
          let sub' = sub {ntfServer, ntfQueueId = Nothing, ntfSubId = Nothing, ntfSubStatus = NASNew}
           in smpServer <$ updateNtfSubscription db sub' (NSASMP NSASmpKey) ts
    incStatByUserId :: NtfServer -> (AgentNtfServerStats -> TVar Int) -> [NtfSubscription] -> AM' ()
    incStatByUserId ntfServer sel ss =
      forM_ (M.assocs userIdsCounts) $ \(userId, count) ->
        atomically $ incNtfServerStat' c userId ntfServer sel count
      where
        userIdsCounts = foldl' (\acc NtfSubscription {userId} -> M.insertWith (+) userId 1 acc) M.empty ss
    -- NSADelete and NSARotate are deprecated, but their processing is kept for legacy db records;
    -- These actions are not batched
    deleteSubs :: [NtfSubscription] -> AM' [NtfSubscription]
    deleteSubs ntfSubs = do
      retrySubs_ <- mapM (runCatching deleteSub) ntfSubs
      pure $ catMaybes retrySubs_
      where
        deleteSub :: NtfSubscription -> AM (Maybe NtfSubscription)
        deleteSub sub@NtfSubscription {smpServer} =
          deleteNtfSub sub $ do
            let sub' = sub {ntfSubId = Nothing, ntfSubStatus = NASOff}
            ts <- liftIO getCurrentTime
            withStore' c $ \db -> updateNtfSubscription db sub' (NSASMP NSASmpDelete) ts
            lift . void $ getNtfSMPWorker True c smpServer
    rotateSubs :: [NtfSubscription] -> AM' [NtfSubscription]
    rotateSubs ntfSubs = do
      retrySubs_ <- mapM (runCatching rotateSub) ntfSubs
      pure $ catMaybes retrySubs_
      where
        rotateSub :: NtfSubscription -> AM (Maybe NtfSubscription)
        rotateSub sub@NtfSubscription {connId} =
          deleteNtfSub sub $ do
            withStore' c $ \db -> deleteNtfSubscription db connId
            ns <- asks ntfSupervisor
            atomically $ writeTBQueue (ntfSubQ ns) (NSCCreate, [connId])
    runCatching :: (NtfSubscription -> AM (Maybe NtfSubscription)) -> NtfSubscription -> AM' (Maybe NtfSubscription)
    runCatching action sub@NtfSubscription {connId} =
      fromRight Nothing
        <$> runExceptT (action sub `catchAllErrors` \e -> workerInternalError c connId (show e) $> Nothing)
    -- deleteNtfSub is only used in NSADelete and NSARotate, so also deprecated
    deleteNtfSub :: NtfSubscription -> AM () -> AM (Maybe NtfSubscription)
    deleteNtfSub sub@NtfSubscription {userId, ntfSubId} continue = case ntfSubId of
      Just nSubId ->
        lift getNtfToken >>= \case
          Just tkn@NtfToken {ntfServer} -> do
            atomically $ incNtfServerStat c userId ntfServer ntfDelAttempts
            tryAllErrors (agentNtfDeleteSubscription c nSubId tkn) >>= \case
              Right _ -> do
                atomically $ incNtfServerStat c userId ntfServer ntfDeleted
                continue'
              Left e
                | temporaryOrHostError e -> pure $ Just sub -- don't continue, retry
                | otherwise -> continue'
          Nothing -> continue'
      _ -> continue'
      where
        continue' = continue $> Nothing -- continue without retry

runNtfSMPWorker :: AgentClient -> SMPServer -> Worker -> AM ()
runNtfSMPWorker c srv Worker {doWork} = forever $ do
  waitForWork doWork
  ExceptT $ agentOperationBracket c AONtfNetwork throwWhenInactive $ runExceptT runNtfSMPOperation
  where
    runNtfSMPOperation :: AM ()
    runNtfSMPOperation = do
      ntfBatchSize <- asks $ ntfBatchSize . config
      withWorkItems c doWork (withStore' c $ \db -> getNextNtfSubSMPActions db srv ntfBatchSize) $ \nextSubs -> do
        logInfo $ "runNtfSMPWorker - length nextSubs = " <> tshow (length nextSubs)
        let (creates, deletes) = splitActions nextSubs
        retrySubActions c creates createNotifierKeys
        retrySubActions c deletes deleteNotifierKeys
    splitActions :: NonEmpty (NtfSubSMPAction, NtfSubscription) -> ([NtfSubscription], [NtfSubscription])
    splitActions = foldr addAction ([], [])
      where
        addAction (cmd, sub) (creates, deletes) = case cmd of
          NSASmpKey -> (sub : creates, deletes)
          NSASmpDelete -> (creates, sub : deletes)
    createNotifierKeys :: [NtfSubscription] -> AM' [NtfSubscription]
    createNotifierKeys ntfSubs =
      getNtfToken >>= \case
        Just NtfToken {ntfTknStatus = NTActive, ntfMode = NMInstant} -> do
          (errs1, subRqKeys) <- prepareQueueSmpKey ntfSubs
          rs <- enableQueuesNtfs c subRqKeys
          let (subRqKeys', errs2, successes) = splitResults rs
              ntfSubs' = map eqnrNtfSub subRqKeys'
              errs2' = map (first (qConnId . eqnrRq)) errs2
          ts <- liftIO getCurrentTime
          (errs3, srvs) <- partitionErrs (qConnId . eqnrRq . fst) successes <$> withStoreBatch' c (\db -> map (storeNtfSubCreds db ts) successes)
          mapM_ (getNtfNTFWorker True c) $ S.fromList srvs
          workerErrors c $ errs1 <> errs2' <> errs3
          pure ntfSubs'
        _ -> do
          let errs = map (\sub -> (ntfSubConnId sub, INTERNAL "NSASmpKey - no active token")) ntfSubs
          workerErrors c errs
          pure []
      where
        prepareQueueSmpKey :: [NtfSubscription] -> AM' ([(ConnId, AgentErrorType)], [EnableQueueNtfReq])
        prepareQueueSmpKey subs = do
          alg <- asks (rcvAuthAlg . config)
          g <- asks random
          partitionErrs ntfSubConnId subs <$> withStoreBatch c (\db -> map (getQueue db alg g) subs)
          where
            getQueue :: DB.Connection -> C.AuthAlg -> TVar ChaChaDRG -> NtfSubscription -> IO (Either AgentErrorType EnableQueueNtfReq)
            getQueue db (C.AuthAlg a) g sub = fmap (first storeError) $ runExceptT $ do
              rq <- ExceptT $ getPrimaryRcvQueue db (ntfSubConnId sub)
              authKeyPair <- atomically $ C.generateAuthKeyPair a g
              rcvNtfKeyPair <- atomically $ C.generateKeyPair g
              pure (EnableQueueNtfReq sub rq authKeyPair rcvNtfKeyPair)
        storeNtfSubCreds :: DB.Connection -> UTCTime -> (EnableQueueNtfReq, (SMP.NotifierId, SMP.RcvNtfPublicDhKey)) -> IO NtfServer
        storeNtfSubCreds db ts (EnableQueueNtfReq {eqnrNtfSub, eqnrAuthKeyPair = (ntfPublicKey, ntfPrivateKey), eqnrRcvKeyPair = (_, pk)}, (notifierId, srvPubDhKey)) = do
          let NtfSubscription {ntfServer} = eqnrNtfSub
              rcvNtfDhSecret = C.dh' srvPubDhKey pk
          setRcvQueueNtfCreds db (ntfSubConnId eqnrNtfSub) $ Just ClientNtfCreds {ntfPublicKey, ntfPrivateKey, notifierId, rcvNtfDhSecret}
          updateNtfSubscription db eqnrNtfSub {ntfQueueId = Just notifierId, ntfSubStatus = NASKey} (NSANtf NSACreate) ts
          pure ntfServer
    deleteNotifierKeys :: [NtfSubscription] -> AM' [NtfSubscription]
    deleteNotifierKeys ntfSubs = do
      (errs1, subRqs) <- partitionErrs ntfSubConnId ntfSubs <$> withStoreBatch c (\db -> map (resetCredsGetQueue db) ntfSubs)
      rs <- disableQueuesNtfs c subRqs
      let (subRqs', errs2, successes) = splitResults rs
          ntfSubs' = map fst subRqs'
          errs2' = map (first (qConnId . snd)) errs2
          disabledRqs = map (snd . fst) successes
      (errs3, _) <- partitionErrs qConnId disabledRqs <$> withStoreBatch' c (\db -> map (deleteSub db) disabledRqs)
      workerErrors c $ errs1 <> errs2' <> errs3
      pure ntfSubs'
      where
        resetCredsGetQueue :: DB.Connection -> NtfSubscription -> IO (Either AgentErrorType DisableQueueNtfReq)
        resetCredsGetQueue db sub@NtfSubscription {connId} = fmap (first storeError) $ runExceptT $ do
          liftIO $ setRcvQueueNtfCreds db connId Nothing
          rq <- ExceptT $ getPrimaryRcvQueue db connId
          pure (sub, rq)
        deleteSub :: DB.Connection -> RcvQueue -> IO ()
        deleteSub db rq = deleteNtfSubscription db (qConnId rq)

retrySubActions :: AgentClient -> [NtfSubscription] -> ([NtfSubscription] -> AM' [NtfSubscription]) -> AM ()
retrySubActions _ [] _ = pure ()
retrySubActions c subs action = do
  v <- newTVarIO subs
  ri <- asks $ reconnectInterval . config
  withRetryInterval ri $ \_ loop -> do
    liftIO $ waitWhileSuspended c
    liftIO $ waitForUserNetwork c
    subs' <- readTVarIO v
    retrySubs <- lift $ action subs'
    unless (null retrySubs) $ do
      atomically $ writeTVar v retrySubs
      retryNetworkLoop c loop

--                                                (temporary errs, other errs, successes)
splitResults :: [(a, Either AgentErrorType r)] -> ([a], [(a, AgentErrorType)], [(a, r)])
splitResults = foldr addRes ([], [], [])
  where
    addRes (a, r_) (as, errs, rs) = case r_ of
      Right r -> (as, errs, (a, r) : rs)
      Left e
        | temporaryOrHostError e -> (a : as, errs, rs)
        | otherwise -> (as, (a, e) : errs, rs)

rescheduleWork :: TMVar () -> UTCTime -> UTCTime -> AM' ()
rescheduleWork doWork ts actionTs = do
  void . atomically $ tryTakeTMVar doWork
  void . forkIO $ do
    liftIO $ threadDelay' $ diffToMicroseconds $ diffUTCTime actionTs ts
    atomically $ hasWorkToDo' doWork

retryNetworkLoop :: AgentClient -> AM () -> AM ()
retryNetworkLoop c loop = do
  atomically $ endAgentOperation c AONtfNetwork
  liftIO $ throwWhenInactive c
  atomically $ beginAgentOperation c AONtfNetwork
  loop

workerErrors :: AgentClient -> [(ConnId, AgentErrorType)] -> AM' ()
workerErrors c connErrs =
  unless (null connErrs) $ do
    void $ withStoreBatch' c (\db -> map (setNullNtfSubscriptionAction db . fst) connErrs)
    notifyErrs c connErrs

workerInternalError :: AgentClient -> ConnId -> String -> AM ()
workerInternalError c connId internalErrStr = do
  withStore' c $ \db -> setNullNtfSubscriptionAction db connId
  notifyInternalError c connId internalErrStr

-- TODO change error
notifyInternalError :: MonadIO m => AgentClient -> ConnId -> String -> m ()
notifyInternalError AgentClient {subQ} connId internalErrStr = do
  logError $ T.pack internalErrStr
  liftIO $ nonBlockingWriteTBQueue subQ ("", connId, AEvt SAEConn $ ERR $ INTERNAL internalErrStr)

notifyInternalError' :: MonadIO m => AgentClient -> String -> m ()
notifyInternalError' c = notifyInternalError c ""
{-# INLINE notifyInternalError' #-}

notifyErrs :: MonadIO m => AgentClient -> [(ConnId, AgentErrorType)] -> m ()
notifyErrs c errs_ = forM_ (L.nonEmpty errs_) $ \errs -> do
  logError $ "notifyErrs: " <> tshow errs
  notifySub c $ ERRS errs

getNtfToken :: AM' (Maybe NtfToken)
getNtfToken = do
  tkn <- asks $ ntfTkn . ntfSupervisor
  readTVarIO tkn

nsUpdateToken :: NtfSupervisor -> NtfToken -> STM ()
nsUpdateToken ns tkn = writeTVar (ntfTkn ns) $ Just tkn

nsRemoveNtfToken :: NtfSupervisor -> STM ()
nsRemoveNtfToken ns = writeTVar (ntfTkn ns) Nothing

sendNtfSubCommand :: NtfSupervisor -> (NtfSupervisorCommand, NonEmpty ConnId) -> IO ()
sendNtfSubCommand ns cmd =
  whenM (hasInstantNotifications ns) $ atomically $ writeTBQueue (ntfSubQ ns) cmd

hasInstantNotifications :: NtfSupervisor -> IO Bool
hasInstantNotifications ns = do
  tkn <- readTVarIO $ ntfTkn ns
  pure $ maybe False instantNotifications tkn

instantNotifications :: NtfToken -> Bool
instantNotifications NtfToken {ntfTknStatus = NTActive, ntfMode = NMInstant} = True
instantNotifications _ = False
{-# INLINE instantNotifications #-}

deleteToken :: AgentClient -> NtfToken -> AM ()
deleteToken c tkn@NtfToken {ntfServer, ntfTokenId, ntfPrivKey} = do
  setToDelete <- withStore' c $ \db -> do
    removeNtfToken db tkn
    case ntfTokenId of
      Just tknId -> addNtfTokenToDelete db ntfServer ntfPrivKey tknId $> True
      Nothing -> pure False
  ns <- asks ntfSupervisor
  atomically $ nsRemoveNtfToken ns
  when setToDelete $ void $ lift $ getNtfTknDelWorker True c ntfServer

runNtfTknDelWorker :: AgentClient -> NtfServer -> Worker -> AM ()
runNtfTknDelWorker c srv Worker {doWork} =
  forever $ do
    waitForWork doWork
    ExceptT $ agentOperationBracket c AONtfNetwork throwWhenInactive $ runExceptT runNtfOperation
  where
    runNtfOperation :: AM ()
    runNtfOperation =
      withWork c doWork (`getNextNtfTokenToDelete` srv) $
        \nextTknToDelete -> do
          logInfo $ "runNtfTknDelWorker, nextTknToDelete " <> tshow nextTknToDelete
          ri <- asks $ reconnectInterval . config
          withRetryInterval ri $ \_ loop -> do
            liftIO $ waitWhileSuspended c
            liftIO $ waitForUserNetwork c
            processTknToDelete nextTknToDelete `catchAllErrors` retryTmpError loop nextTknToDelete
    retryTmpError :: AM () -> NtfTokenToDelete -> AgentErrorType -> AM ()
    retryTmpError loop (tknDbId, _, _) e = do
      logError $ "ntf tkn del error: " <> tshow e
      if temporaryOrHostError e
        then retryNetworkLoop c loop
        else do
          withStore' c $ \db -> deleteNtfTokenToDelete db tknDbId
          notifyInternalError' c (show e)
    processTknToDelete :: NtfTokenToDelete -> AM ()
    processTknToDelete (tknDbId, ntfPrivKey, tknId) = do
      agentNtfDeleteToken c NRMBackground srv ntfPrivKey tknId
      withStore' c $ \db -> deleteNtfTokenToDelete db tknDbId

closeNtfSupervisor :: NtfSupervisor -> IO ()
closeNtfSupervisor ns = do
  stopWorkers $ ntfWorkers ns
  stopWorkers $ ntfSMPWorkers ns
  stopWorkers $ ntfTknDelWorkers ns
  where
    stopWorkers workers = atomically (swapTVar workers M.empty) >>= mapM_ (liftIO . cancelWorker)

getNtfServer :: AgentClient -> AM' (Maybe NtfServer)
getNtfServer c = do
  ntfServers <- readTVarIO $ ntfServers c
  case ntfServers of
    [] -> pure Nothing
    [srv] -> pure $ Just srv
    servers -> do
      gen <- asks randomServer
      atomically . stateTVar gen $
        first (Just . (servers !!)) . randomR (0, length servers - 1)