Merge branch 'master' into ntf-storage

Author: epoberezkin

simplexmq.cabal

@@ -1,7 +1,7 @@
 cabal-version: 1.12
 
 name:           simplexmq
-version:        6.4.0.1
+version:        6.4.0.2
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and

src/Simplex/Messaging/Agent.hs

@@ -439,7 +439,7 @@ subscribeConnections c = withAgentEnv c . subscribeConnections' c
 {-# INLINE subscribeConnections #-}
 
 -- | Get messages for connections (GET commands)
-getConnectionMessages :: AgentClient -> NonEmpty ConnId -> IO (NonEmpty (Maybe SMPMsgMeta))
+getConnectionMessages :: AgentClient -> NonEmpty ConnMsgReq -> IO (NonEmpty (Either AgentErrorType (Maybe SMPMsgMeta)))
 getConnectionMessages c = withAgentEnv' c . getConnectionMessages' c
 {-# INLINE getConnectionMessages #-}
 
@@ -1276,24 +1276,26 @@ resubscribeConnections' c connIds = do
   -- union is left-biased, so results returned by subscribeConnections' take precedence
   (`M.union` r) <$> subscribeConnections' c connIds'
 
-getConnectionMessages' :: AgentClient -> NonEmpty ConnId -> AM' (NonEmpty (Maybe SMPMsgMeta))
-getConnectionMessages' c = mapM getMsg
+-- requesting messages sequentially, to reduce memory usage
+getConnectionMessages' :: AgentClient -> NonEmpty ConnMsgReq -> AM' (NonEmpty (Either AgentErrorType (Maybe SMPMsgMeta)))
+getConnectionMessages' c = mapM $ tryAgentError' . getConnectionMessage
   where
-    getMsg :: ConnId -> AM' (Maybe SMPMsgMeta)
-    getMsg connId =
-      getConnectionMessage connId `catchAgentError'` \e -> do
-        logError $ "Error loading message: " <> tshow e
-        pure Nothing
-    getConnectionMessage :: ConnId -> AM (Maybe SMPMsgMeta)
-    getConnectionMessage connId = do
+    getConnectionMessage :: ConnMsgReq -> AM (Maybe SMPMsgMeta)
+    getConnectionMessage (ConnMsgReq connId dbQueueId msgTs_) = do
       whenM (atomically $ hasActiveSubscription c connId) . throwE $ CMD PROHIBITED "getConnectionMessage: subscribed"
       SomeConn _ conn <- withStore c (`getConn` connId)
-      case conn of
-        DuplexConnection _ (rq :| _) _ -> getQueueMessage c rq
-        RcvConnection _ rq -> getQueueMessage c rq
-        ContactConnection _ rq -> getQueueMessage c rq
+      rq <- case conn of
+        DuplexConnection _ (rq :| _) _ -> pure rq
+        RcvConnection _ rq -> pure rq
+        ContactConnection _ rq -> pure rq
         SndConnection _ _ -> throwE $ CONN SIMPLEX
         NewConnection _ -> throwE $ CMD PROHIBITED "getConnectionMessage: NewConnection"
+      msg_ <- getQueueMessage c rq `catchAgentError` \e -> atomically (releaseGetLock c rq) >> throwError e
+      when (isNothing msg_) $ do
+        atomically $ releaseGetLock c rq
+        forM_ msgTs_ $ \msgTs -> withStore' c $ \db -> setLastBrokerTs db connId (DBQueueId dbQueueId) msgTs
+      pure msg_
+{-# INLINE getConnectionMessages' #-}
 
 getNotificationConns' :: AgentClient -> C.CbNonce -> ByteString -> AM (NonEmpty NotificationInfo)
 getNotificationConns' c nonce encNtfInfo =
@@ -1308,25 +1310,26 @@ getNotificationConns' c nonce encNtfInfo =
               lastNtfInfo = Just . fst <$$> getNtfInfo db lastNtf
            in initNtfInfos <> [lastNtfInfo]
       let (errs, ntfInfos_) = partitionEithers rs
-      logError $ "Error(s) loading notifications: " <> tshow errs
+      unless (null errs) $ logError $ "Error(s) loading notifications: " <> tshow errs
       case L.nonEmpty $ catMaybes ntfInfos_ of
         Just r -> pure r
         Nothing -> throwE $ INTERNAL "getNotificationConns: couldn't get conn info"
     _ -> throwE $ CMD PROHIBITED "getNotificationConns"
   where
     getNtfInfo :: DB.Connection -> PNMessageData -> IO (Either AgentErrorType (NotificationInfo, Maybe UTCTime))
     getNtfInfo db PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta} = runExceptT $ do
-      (ntfConnId, rcvNtfDhSecret, lastBrokerTs_) <- liftError' storeError $ getNtfRcvQueue db smpQueue
+      (ntfConnId, ntfDbQueueId, rcvNtfDhSecret, lastBrokerTs_) <- liftError' storeError $ getNtfRcvQueue db smpQueue
       let ntfMsgMeta = eitherToMaybe $ smpDecode =<< first show (C.cbDecrypt rcvNtfDhSecret nmsgNonce encNMsgMeta)
-          ntfInfo = NotificationInfo {ntfConnId, ntfTs, ntfMsgMeta}
+          ntfInfo = NotificationInfo {ntfConnId, ntfDbQueueId, ntfTs, ntfMsgMeta}
       pure (ntfInfo, lastBrokerTs_)
     getInitNtfInfo :: DB.Connection -> PNMessageData -> IO (Either AgentErrorType (Maybe NotificationInfo))
     getInitNtfInfo db msgData = runExceptT $ do
-      (nftInfo, lastBrokerTs_) <- ExceptT $ getNtfInfo db msgData
-      pure $ case (ntfMsgMeta nftInfo, lastBrokerTs_) of
-        (Just SMP.NMsgMeta {msgTs}, Just lastBrokerTs)
-          | systemToUTCTime msgTs > lastBrokerTs -> Just nftInfo
+      (ntfInfo, lastBrokerTs_) <- ExceptT $ getNtfInfo db msgData
+      pure $ case ntfMsgMeta ntfInfo of
+        Just SMP.NMsgMeta {msgTs}
+          | maybe True (systemToUTCTime msgTs >) lastBrokerTs_ -> Just ntfInfo
         _ -> Nothing
+{-# INLINE getNotificationConns' #-}
 
 -- | Send message to the connection (SEND command) in Reader monad
 sendMessage' :: AgentClient -> ConnId -> PQEncryption -> MsgFlags -> MsgBody -> AM (AgentMsgId, PQEncryption)

src/Simplex/Messaging/Agent/Client.hs

@@ -326,6 +326,7 @@ data AgentClient = AgentClient
     xftpServers :: TMap UserId (UserServers 'PXFTP),
     xftpClients :: TMap XFTPTransportSession XFTPClientVar,
     useNetworkConfig :: TVar (NetworkConfig, NetworkConfig), -- (slow, fast) networks
+    presetSMPDomains :: [HostName],
     userNetworkInfo :: TVar UserNetworkInfo,
     userNetworkUpdated :: TVar (Maybe UTCTime),
     subscrConns :: TVar (Set ConnId),
@@ -478,7 +479,7 @@ data UserNetworkType = UNNone | UNCellular | UNWifi | UNEthernet | UNOther
 
 -- | Creates an SMP agent client instance that receives commands and sends responses via 'TBQueue's.
 newAgentClient :: Int -> InitialAgentServers -> UTCTime -> Env -> IO AgentClient
-newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} currentTs agentEnv = do
+newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg, presetDomains} currentTs agentEnv = do
   let cfg = config agentEnv
       qSize = tbqSize cfg
   proxySessTs <- newTVarIO =<< getCurrentTime
@@ -532,6 +533,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} currentTs a
         xftpServers,
         xftpClients,
         useNetworkConfig,
+        presetSMPDomains = presetDomains,
         userNetworkInfo,
         userNetworkUpdated,
         subscrConns,
@@ -690,7 +692,7 @@ smpConnectClient c@AgentClient {smpClients, msgQ, proxySessTs} tSess@(_, srv, _)
       env <- ask
       liftError (protocolClientError SMP $ B.unpack $ strEncode srv) $ do
         ts <- readTVarIO proxySessTs
-        smp <- ExceptT $ getProtocolClient g tSess cfg (Just msgQ) ts $ smpClientDisconnected c tSess env v' prs
+        smp <- ExceptT $ getProtocolClient g tSess cfg (presetSMPDomains c) (Just msgQ) ts $ smpClientDisconnected c tSess env v' prs
         pure SMPConnectedClient {connectedClient = smp, proxiedRelays = prs}
 
 smpClientDisconnected :: AgentClient -> SMPTransportSession -> Env -> SMPClientVar -> TMap SMPServer ProxiedRelayVar -> SMPClient -> IO ()
@@ -793,7 +795,7 @@ getNtfServerClient c@AgentClient {active, ntfClients, workerSeq, proxySessTs} tS
       g <- asks random
       ts <- readTVarIO proxySessTs
       liftError' (protocolClientError NTF $ B.unpack $ strEncode srv) $
-        getProtocolClient g tSess cfg Nothing ts $
+        getProtocolClient g tSess cfg [] Nothing ts $
           clientDisconnected v
 
     clientDisconnected :: NtfClientVar -> NtfClient -> IO ()
@@ -1225,7 +1227,7 @@ runSMPServerTest c userId (ProtoServerWithAuth srv auth) = do
   liftIO $ do
     let tSess = (userId, srv, Nothing)
     ts <- readTVarIO $ proxySessTs c
-    getProtocolClient g tSess cfg Nothing ts (\_ -> pure ()) >>= \case
+    getProtocolClient g tSess cfg (presetSMPDomains c) Nothing ts (\_ -> pure ()) >>= \case
       Right smp -> do
         rKeys@(_, rpKey) <- atomically $ C.generateAuthKeyPair ra g
         (sKey, spKey) <- atomically $ C.generateAuthKeyPair sa g
@@ -1302,7 +1304,7 @@ runNTFServerTest c userId (ProtoServerWithAuth srv _) = do
   liftIO $ do
     let tSess = (userId, srv, Nothing)
     ts <- readTVarIO $ proxySessTs c
-    getProtocolClient g tSess cfg Nothing ts (\_ -> pure ()) >>= \case
+    getProtocolClient g tSess cfg [] Nothing ts (\_ -> pure ()) >>= \case
       Right ntf -> do
         (nKey, npKey) <- atomically $ C.generateAuthKeyPair a g
         (dhKey, _) <- atomically $ C.generateKeyPair g
@@ -1652,6 +1654,7 @@ getQueueMessage c rq@RcvQueue {server, rcvId, rcvPrivateKey} = do
           l <- maybe (newTMVar ()) pure l_
           takeTMVar l
           pure $ Just l
+{-# INLINE getQueueMessage #-}
 
 decryptSMPMessage :: RcvQueue -> SMP.RcvMessage -> AM SMP.ClientRcvMsgBody
 decryptSMPMessage rq SMP.RcvMessage {msgId, msgBody = SMP.EncRcvMsgBody body} =
@@ -1741,10 +1744,12 @@ sendAck c rq@RcvQueue {rcvId, rcvPrivateKey} msgId =
 hasGetLock :: AgentClient -> RcvQueue -> IO Bool
 hasGetLock c RcvQueue {server, rcvId} =
   TM.memberIO (server, rcvId) $ getMsgLocks c
+{-# INLINE hasGetLock #-}
 
 releaseGetLock :: AgentClient -> RcvQueue -> STM ()
 releaseGetLock c RcvQueue {server, rcvId} =
   TM.lookup (server, rcvId) (getMsgLocks c) >>= mapM_ (`tryPutTMVar` ())
+{-# INLINE releaseGetLock #-}
 
 suspendQueue :: AgentClient -> RcvQueue -> AM ()
 suspendQueue c rq@RcvQueue {rcvId, rcvPrivateKey} =

src/Simplex/Messaging/Agent/Env/SQLite.hs

@@ -96,7 +96,8 @@ data InitialAgentServers = InitialAgentServers
   { smp :: Map UserId (NonEmpty (ServerCfg 'PSMP)),
     ntf :: [NtfServer],
     xftp :: Map UserId (NonEmpty (ServerCfg 'PXFTP)),
-    netCfg :: NetworkConfig
+    netCfg :: NetworkConfig,
+    presetDomains :: [HostName]
   }
 
 data ServerCfg p = ServerCfg

src/Simplex/Messaging/Agent/Protocol.hs

@@ -147,6 +147,7 @@ module Simplex.Messaging.Agent.Protocol
     AgentMsgId,
     NotificationsMode (..),
     NotificationInfo (..),
+    ConnMsgReq (..),
 
     -- * Encode/decode
     serializeCommand,
@@ -678,11 +679,21 @@ instance FromField NotificationsMode where fromField = blobFieldDecoder $ parseA
 
 data NotificationInfo = NotificationInfo
   { ntfConnId :: ConnId,
+    ntfDbQueueId :: Int64,
     ntfTs :: SystemTime,
+    -- Nothing means that the message failed to decrypt or to decode,
+    -- we can still show event notification
     ntfMsgMeta :: Maybe NMsgMeta
   }
   deriving (Show)
 
+data ConnMsgReq = ConnMsgReq
+  { msgConnId :: ConnId,
+    msgDbQueueId :: Int64,
+    msgTs :: Maybe UTCTime
+  }
+  deriving (Show)
+
 data ConnectionMode = CMInvitation | CMContact
   deriving (Eq, Show)
 

src/Simplex/Messaging/Agent/Store/AgentStore.hs

@@ -98,6 +98,7 @@ module Simplex.Messaging.Agent.Store.AgentStore
     -- Messages
     updateRcvIds,
     createRcvMsg,
+    setLastBrokerTs,
     updateRcvMsgHash,
     createSndMsgBody,
     updateSndIds,
@@ -855,7 +856,11 @@ createRcvMsg db connId rq@RcvQueue {dbQueueId} rcvMsgData@RcvMsgData {msgMeta =
   insertRcvMsgBase_ db connId rcvMsgData
   insertRcvMsgDetails_ db connId rq rcvMsgData
   updateRcvMsgHash db connId sndMsgId internalRcvId internalHash
-  DB.execute db "UPDATE rcv_queues SET last_broker_ts = ? WHERE conn_id = ? AND rcv_queue_id = ?" (brokerTs, connId, dbQueueId)
+  setLastBrokerTs db connId dbQueueId brokerTs
+
+setLastBrokerTs :: DB.Connection -> ConnId -> DBQueueId 'QSStored -> UTCTime -> IO ()
+setLastBrokerTs db connId dbQueueId brokerTs =
+  DB.execute db "UPDATE rcv_queues SET last_broker_ts = ? WHERE conn_id = ? AND rcv_queue_id = ? AND (last_broker_ts IS NULL OR last_broker_ts < ?)" (brokerTs, connId, dbQueueId, brokerTs)
 
 createSndMsgBody :: DB.Connection -> AMessage -> IO Int64
 createSndMsgBody db aMessage =
@@ -1781,19 +1786,19 @@ getActiveNtfToken db =
           ntfMode = fromMaybe NMPeriodic ntfMode_
        in NtfToken {deviceToken = DeviceToken provider dt, ntfServer, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhKeys, ntfDhSecret, ntfTknStatus, ntfTknAction, ntfMode}
 
-getNtfRcvQueue :: DB.Connection -> SMPQueueNtf -> IO (Either StoreError (ConnId, RcvNtfDhSecret, Maybe UTCTime))
+getNtfRcvQueue :: DB.Connection -> SMPQueueNtf -> IO (Either StoreError (ConnId, Int64, RcvNtfDhSecret, Maybe UTCTime))
 getNtfRcvQueue db SMPQueueNtf {smpServer = (SMPServer host port _), notifierId} =
   firstRow' res SEConnNotFound $
     DB.query
       db
       [sql|
-        SELECT conn_id, rcv_ntf_dh_secret, last_broker_ts
+        SELECT conn_id, rcv_queue_id, rcv_ntf_dh_secret, last_broker_ts
         FROM rcv_queues
         WHERE host = ? AND port = ? AND ntf_id = ? AND deleted = 0
       |]
       (host, port, notifierId)
   where
-    res (connId, Just rcvNtfDhSecret, lastBrokerTs_) = Right (connId, rcvNtfDhSecret, lastBrokerTs_)
+    res (connId, dbQueueId, Just rcvNtfDhSecret, lastBrokerTs_) = Right (connId, dbQueueId, rcvNtfDhSecret, lastBrokerTs_)
     res _ = Left SEConnNotFound
 
 setConnectionNtfs :: DB.Connection -> ConnId -> Bool -> IO ()

src/Simplex/Messaging/Client.hs

@@ -84,6 +84,7 @@ module Simplex.Messaging.Client
     SocksMode (..),
     SMPProxyMode (..),
     SMPProxyFallback (..),
+    SMPWebPortServers (..),
     defaultClientConfig,
     defaultSMPClientConfig,
     defaultNetworkConfig,
@@ -129,7 +130,7 @@ import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Base64 as B64
 import Data.Functor (($>))
 import Data.Int (Int64)
-import Data.List (find)
+import Data.List (find, isSuffixOf)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
 import Data.Maybe (catMaybes, fromMaybe)
@@ -138,7 +139,7 @@ import qualified Data.Text as T
 import Data.Time.Clock (UTCTime (..), diffUTCTime, getCurrentTime)
 import qualified Data.X509 as X
 import qualified Data.X509.Validation as XV
-import Network.Socket (ServiceName)
+import Network.Socket (HostName, ServiceName)
 import Network.Socks5 (SocksCredentials (..))
 import Numeric.Natural
 import qualified Simplex.Messaging.Crypto as C
@@ -291,7 +292,7 @@ data NetworkConfig = NetworkConfig
     -- | Fallback to direct connection when destination SMP relay does not support SMP proxy protocol extensions
     smpProxyFallback :: SMPProxyFallback,
     -- | use web port 443 for SMP protocol
-    smpWebPort :: Bool,
+    smpWebPortServers :: SMPWebPortServers,
     -- | timeout for the initial client TCP/TLS connection (microseconds)
     tcpConnectTimeout :: Int,
     -- | timeout of protocol commands (microseconds)
@@ -327,6 +328,12 @@ data SMPProxyFallback
   | SPFProhibit -- prohibit direct connection to destination relay.
   deriving (Eq, Show)
 
+data SMPWebPortServers
+  = SWPAll
+  | SWPPreset
+  | SWPOff
+  deriving (Eq, Show)
+
 instance StrEncoding SMPProxyMode where
   strEncode = \case
     SPMAlways -> "always"
@@ -353,6 +360,18 @@ instance StrEncoding SMPProxyFallback where
       "no" -> pure SPFProhibit
       _ -> fail "Invalid SMP proxy fallback mode"
 
+instance StrEncoding SMPWebPortServers where
+  strEncode = \case
+    SWPAll -> "all"
+    SWPPreset -> "preset"
+    SWPOff -> "off"
+  strP =
+    A.takeTill (== ' ') >>= \case
+      "all" -> pure SWPAll
+      "preset" -> pure SWPPreset
+      "off" -> pure SWPOff
+      _ -> fail "Invalid SMP wep port setting"
+
 defaultNetworkConfig :: NetworkConfig
 defaultNetworkConfig =
   NetworkConfig
@@ -363,7 +382,7 @@ defaultNetworkConfig =
       sessionMode = TSMSession,
       smpProxyMode = SPMNever,
       smpProxyFallback = SPFAllow,
-      smpWebPort = False,
+      smpWebPortServers = SWPPreset,
       tcpConnectTimeout = defaultTcpConnectTimeout,
       tcpTimeout = 15_000_000,
       tcpTimeoutPerKb = 5_000,
@@ -498,15 +517,15 @@ type TransportSession msg = (UserId, ProtoServer msg, Maybe ByteString)
 --
 -- A single queue can be used for multiple 'SMPClient' instances,
 -- as 'SMPServerTransmission' includes server information.
-getProtocolClient :: forall v err msg. Protocol v err msg => TVar ChaChaDRG -> TransportSession msg -> ProtocolClientConfig v -> Maybe (TBQueue (ServerTransmissionBatch v err msg)) -> UTCTime -> (ProtocolClient v err msg -> IO ()) -> IO (Either (ProtocolClientError err) (ProtocolClient v err msg))
-getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, networkConfig, clientALPN, serverVRange, agreeSecret, proxyServer, useSNI} msgQ proxySessTs disconnected = do
+getProtocolClient :: forall v err msg. Protocol v err msg => TVar ChaChaDRG -> TransportSession msg -> ProtocolClientConfig v -> [HostName] -> Maybe (TBQueue (ServerTransmissionBatch v err msg)) -> UTCTime -> (ProtocolClient v err msg -> IO ()) -> IO (Either (ProtocolClientError err) (ProtocolClient v err msg))
+getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, networkConfig, clientALPN, serverVRange, agreeSecret, proxyServer, useSNI} presetDomains msgQ proxySessTs disconnected = do
   case chooseTransportHost networkConfig (host srv) of
     Right useHost ->
       (getCurrentTime >>= mkProtocolClient useHost >>= runClient useTransport useHost)
         `catch` \(e :: IOException) -> pure . Left $ PCEIOError e
     Left e -> pure $ Left e
   where
-    NetworkConfig {smpWebPort, tcpConnectTimeout, tcpTimeout, smpPingInterval} = networkConfig
+    NetworkConfig {smpWebPortServers, tcpConnectTimeout, tcpTimeout, smpPingInterval} = networkConfig
     mkProtocolClient :: TransportHost -> UTCTime -> IO (PClient v err msg)
     mkProtocolClient transportHost ts = do
       connected <- newTVarIO False
@@ -554,6 +573,13 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
         SPSMP | smpWebPort -> ("443", transport @TLS)
         _ -> defaultTransport cfg
       p -> (p, transport @TLS)
+      where
+        smpWebPort = case smpWebPortServers of
+          SWPAll -> True
+          SWPPreset -> case srv of
+            ProtocolServer {host = THDomainName h :| _} -> any (`isSuffixOf` h) presetDomains
+            _ -> False
+          SWPOff -> False
 
     client :: forall c. Transport c => TProxy c -> PClient v err msg -> TMVar (Either (ProtocolClientError err) (ProtocolClient v err msg)) -> c -> IO ()
     client _ c cVar h = do
@@ -775,6 +801,7 @@ getSMPMessage c rpKey rId =
     OK -> pure Nothing
     cmd@(MSG msg) -> liftIO (writeSMPMessage c rId cmd) $> Just msg
     r -> throwE $ unexpectedResponse r
+{-# INLINE getSMPMessage #-}
 
 -- | Subscribe to the SMP queue notifications.
 --
@@ -1262,6 +1289,8 @@ $(J.deriveJSON (enumJSON $ dropPrefix "SPM") ''SMPProxyMode)
 
 $(J.deriveJSON (enumJSON $ dropPrefix "SPF") ''SMPProxyFallback)
 
+$(J.deriveJSON (enumJSON $ dropPrefix "SWP") ''SMPWebPortServers)
+
 $(J.deriveJSON defaultJSON ''NetworkConfig)
 
 $(J.deriveJSON (sumTypeJSON $ dropPrefix "Proxy") ''ProxyClientError)

src/Simplex/Messaging/Client/Agent.hs

@@ -198,7 +198,7 @@ isOwnServer SMPClientAgent {agentCfg} ProtocolServer {host} =
 -- | Run an SMP client for SMPClientVar
 connectClient :: SMPClientAgent -> SMPServer -> SMPClientVar -> IO (Either SMPClientError SMPClient)
 connectClient ca@SMPClientAgent {agentCfg, smpClients, smpSessions, msgQ, randomDrg, startedAt} srv v =
-  getProtocolClient randomDrg (1, srv, Nothing) (smpCfg agentCfg) (Just msgQ) startedAt clientDisconnected
+  getProtocolClient randomDrg (1, srv, Nothing) (smpCfg agentCfg) [] (Just msgQ) startedAt clientDisconnected
   where
     clientDisconnected :: SMPClient -> IO ()
     clientDisconnected smp = do