6.4.0.6

* smp server: allow getting and deleting short links for the old contact queues

* fix verifaction of legacy contact queues

* test

simplexmq.cabal

@@ -1,7 +1,7 @@
 cabal-version: 1.12
 
 name:           simplexmq
-version:        6.4.0.4
+version:        6.4.0.6
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -135,6 +135,7 @@ library
       Simplex.Messaging.Server.QueueStore.QueueInfo
       Simplex.Messaging.ServiceScheme
       Simplex.Messaging.Session
+      Simplex.Messaging.Agent.Store.Entity
       Simplex.Messaging.TMap
       Simplex.Messaging.Transport
       Simplex.Messaging.Transport.Buffer
@@ -308,6 +309,7 @@ library
     , network-transport ==0.5.6
     , network-udp ==0.0.*
     , random >=1.1 && <1.3
+    , scientific ==0.3.7.*
     , simple-logger ==0.1.*
     , socks ==0.6.*
     , stm ==2.5.*
@@ -519,6 +521,7 @@ test-suite simplexmq-test
     , generic-random ==1.5.*
     , hashable
     , hspec ==2.11.*
+    , hspec-core ==2.11.*
     , http-client
     , http-types
     , http2

src/Simplex/FileTransfer/Transport.hs

@@ -102,7 +102,7 @@ supportedFileServerVRange :: VersionRangeXFTP
 supportedFileServerVRange = mkVersionRange initialXFTPVersion currentXFTPVersion
 
 -- XFTP protocol does not use this handshake method
-xftpClientHandshakeStub :: c -> Maybe C.KeyPairX25519 -> C.KeyHash -> VersionRangeXFTP -> Bool -> ExceptT TransportError IO (THandle XFTPVersion c 'TClient)
+xftpClientHandshakeStub :: c 'TClient -> Maybe C.KeyPairX25519 -> C.KeyHash -> VersionRangeXFTP -> Bool -> ExceptT TransportError IO (THandle XFTPVersion c 'TClient)
 xftpClientHandshakeStub _c _ks _keyHash _xftpVRange _proxyServer = throwE TEVersion
 
 supportedXFTPhandshakes :: [ALPN]

src/Simplex/Messaging/Agent.hs

@@ -216,6 +216,7 @@ import Simplex.Messaging.Protocol
   )
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.ServiceScheme (ServiceScheme (..))
+import Simplex.Messaging.Agent.Store.Entity
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport (SMPVersion)
 import Simplex.Messaging.Util
@@ -371,8 +372,8 @@ createConnection c userId enableNtfs = withAgentEnv c .::. newConn c userId enab
 {-# INLINE createConnection #-}
 
 -- | Create or update user's contact connection short link
-setContactShortLink :: AgentClient -> ConnId -> ConnInfo -> AE (ConnShortLink 'CMContact)
-setContactShortLink c = withAgentEnv c .: setContactShortLink' c
+setContactShortLink :: AgentClient -> ConnId -> ConnInfo -> Maybe CRClientData -> AE (ConnShortLink 'CMContact)
+setContactShortLink c = withAgentEnv c .:. setContactShortLink' c
 {-# INLINE setContactShortLink #-}
 
 deleteContactShortLink :: AgentClient -> ConnId -> AE ()
@@ -832,8 +833,8 @@ newConn c userId enableNtfs cMode userData_ clientData pqInitKeys subMode = do
   (connId,) <$> newRcvConnSrv c userId connId enableNtfs cMode userData_ clientData pqInitKeys subMode srv
     `catchE` \e -> withStore' c (`deleteConnRecord` connId) >> throwE e
 
-setContactShortLink' :: AgentClient -> ConnId -> ConnInfo -> AM (ConnShortLink 'CMContact)
-setContactShortLink' c connId userData = 
+setContactShortLink' :: AgentClient -> ConnId -> ConnInfo -> Maybe CRClientData -> AM (ConnShortLink 'CMContact)
+setContactShortLink' c connId userData clientData =
   withConnLock c connId "setContactShortLink" $
     withStore c (`getConn` connId) >>= \case
       SomeConn _ (ContactConnection _ rq) -> do
@@ -855,7 +856,7 @@ setContactShortLink' c connId userData =
         Nothing -> do
           sigKeys@(_, privSigKey) <- atomically $ C.generateKeyPair @'C.Ed25519 g
           let qUri = SMPQueueUri vr $ SMPQueueAddress server sndId (C.publicKey e2ePrivKey) (Just QMContact)
-              connReq = CRContactUri $ ConnReqUriData SSSimplex smpAgentVRange [qUri] Nothing
+              connReq = CRContactUri $ ConnReqUriData SSSimplex smpAgentVRange [qUri] clientData
               (linkKey, linkData) = SL.encodeSignLinkData sigKeys smpAgentVRange connReq userData
               (linkId, k) = SL.contactShortLinkKdf linkKey
           srvData <- liftError id $ SL.encryptLinkData g k linkData
@@ -934,7 +935,7 @@ newRcvConnSrv c userId connId enableNtfs cMode userData_ clientData pqInitKeys s
     createRcvQueue nonce_ qd e2eKeys = do
       AgentConfig {smpClientVRange = vr} <- asks config
       -- TODO [notifications] send correct NTF credentials here
-      -- let ntfCreds_ = Nothing 
+      -- let ntfCreds_ = Nothing
       (rq, qUri, tSess, sessId) <- newRcvQueue_ c userId connId srvWithAuth vr qd subMode nonce_ e2eKeys `catchAgentError` \e -> liftIO (print e) >> throwE e
       atomically $ incSMPServerStat c userId srv connCreated
       rq' <- withStore c $ \db -> updateNewConnRcv db connId rq
@@ -1122,7 +1123,7 @@ joinConnSrv c userId connId enableNtfs cReqUri@CRContactUri {} cInfo pqSup subMo
     Nothing -> throwE $ AGENT A_VERSION
 
 delInvSL :: AgentClient -> ConnId -> SMPServerWithAuth -> SMP.LinkId -> AM ()
-delInvSL c connId srv lnkId = 
+delInvSL c connId srv lnkId =
   withStore' c (\db -> deleteInvShortLink db (protoServer srv) lnkId) `catchE` \e ->
     liftIO $ nonBlockingWriteTBQueue (subQ c) ("", connId, AEvt SAEConn (ERR $ INTERNAL $ "error deleting short link " <> show e))
 
@@ -1293,7 +1294,7 @@ getConnectionMessages' c = mapM $ tryAgentError' . getConnectionMessage
       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
+        forM_ msgTs_ $ \msgTs -> withStore' c $ \db -> setLastBrokerTs db connId (DBEntityId dbQueueId) msgTs
       pure msg_
 {-# INLINE getConnectionMessages' #-}
 
@@ -1910,7 +1911,7 @@ switchConnection' c connId =
       _ -> throwE $ CMD PROHIBITED "switchConnection: not duplex"
 
 switchDuplexConnection :: AgentClient -> Connection 'CDuplex -> RcvQueue -> AM ConnectionStats
-switchDuplexConnection c (DuplexConnection cData@ConnData {connId, userId} rqs sqs) rq@RcvQueue {server, dbQueueId = DBQueueId dbQueueId, sndId} = do
+switchDuplexConnection c (DuplexConnection cData@ConnData {connId, userId} rqs sqs) rq@RcvQueue {server, dbQueueId = DBEntityId dbQueueId, sndId} = do
   checkRQSwchStatus rq RSSwitchStarted
   clientVRange <- asks $ smpClientVRange . config
   -- try to get the server that is different from all queues, or at least from the primary rcv queue
@@ -2940,7 +2941,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
               Just qInfo@(Compatible sqInfo@SMPQueueInfo {queueAddress}) ->
                 case (findQ (qAddress sqInfo) sqs, findQ addr sqs) of
                   (Just _, _) -> qError "QADD: queue address is already used in connection"
-                  (_, Just sq@SndQueue {dbQueueId = DBQueueId dbQueueId}) -> do
+                  (_, Just sq@SndQueue {dbQueueId = DBEntityId dbQueueId}) -> do
                     let (delSqs, keepSqs) = L.partition ((Just dbQueueId ==) . dbReplaceQId) sqs
                     case L.nonEmpty keepSqs of
                       Just sqs' -> do
@@ -3278,7 +3279,7 @@ newSndQueue userId connId (Compatible (SMPQueueInfo smpClientVersion SMPQueueAdd
             e2ePubKey = Just e2ePubKey,
             -- setting status to Secured prevents SKEY when queue was already secured with LKEY
             status = if isJust sndKeys_ then Secured else New,
-            dbQueueId = DBNewQueue,
+            dbQueueId = DBNewEntity,
             primary = True,
             dbReplaceQueueId = Nothing,
             sndSwchStatus = Nothing,

src/Simplex/Messaging/Agent/Client.hs

@@ -278,6 +278,7 @@ import Simplex.Messaging.Protocol
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server.QueueStore.QueueInfo
 import Simplex.Messaging.Session
+import Simplex.Messaging.Agent.Store.Entity
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport (SMPVersion, SessionId, THandleParams (sessionId, thVersion), TransportError (..), TransportPeer (..), sndAuthKeySMPVersion, shortLinksSMPVersion)
@@ -1083,7 +1084,7 @@ sendOrProxySMPCommand ::
   UserId ->
   SMPServer ->
   ConnId -> -- session entity ID, for short links LinkId is used
-  ByteString -> 
+  ByteString ->
   SMP.EntityId -> -- sender or link ID
   (SMPClient -> ProxiedRelay -> ExceptT SMPClientError IO (Either ProxyClientError a)) ->
   (SMPClient -> ExceptT SMPClientError IO a) ->
@@ -1395,7 +1396,7 @@ newRcvQueue_ c userId connId (ProtoServerWithAuth srv auth) vRange cqrd subMode
             queueMode,
             shortLink,
             status = New,
-            dbQueueId = DBNewQueue,
+            dbQueueId = DBNewEntity,
             primary = True,
             dbReplaceQueueId = Nothing,
             rcvSwchStatus = Nothing,
@@ -1408,7 +1409,7 @@ newRcvQueue_ c userId connId (ProtoServerWithAuth srv auth) vRange cqrd subMode
   where
     mkShortLinkCreds :: (THandleParams SMPVersion 'TClient, QueueIdsKeys) -> AM (Maybe ShortLinkCreds)
     mkShortLinkCreds (thParams', QIK {sndId, queueMode, linkId}) = case (cqrd, queueMode) of
-      (CQRMessaging ld, Just QMMessaging) -> 
+      (CQRMessaging ld, Just QMMessaging) ->
         withLinkData ld $ \lnkId CQRData {linkKey, privSigKey, srvReq = (sndId', d)} ->
           if sndId == sndId'
             then pure $ Just $ ShortLinkCreds lnkId linkKey privSigKey (fst d)

src/Simplex/Messaging/Agent/Store.hs

@@ -52,30 +52,19 @@ import Simplex.Messaging.Protocol
     VersionSMPC,
   )
 import qualified Simplex.Messaging.Protocol as SMP
+import Simplex.Messaging.Agent.Store.Entity
 
 createStore :: DBOpts -> MigrationConfirmation -> IO (Either MigrationError DBStore)
 createStore dbOpts = createDBStore dbOpts appMigrations
 
 -- * Queue types
 
-data QueueStored = QSStored | QSNew
+type RcvQueue = StoredRcvQueue 'DBStored
 
-data SQueueStored (q :: QueueStored) where
-  SQSStored :: SQueueStored 'QSStored
-  SQSNew :: SQueueStored 'QSNew
-
-data DBQueueId (q :: QueueStored) where
-  DBQueueId :: Int64 -> DBQueueId 'QSStored
-  DBNewQueue :: DBQueueId 'QSNew
-
-deriving instance Show (DBQueueId q)
-
-type RcvQueue = StoredRcvQueue 'QSStored
-
-type NewRcvQueue = StoredRcvQueue 'QSNew
+type NewRcvQueue = StoredRcvQueue 'DBNew
 
 -- | A receive queue. SMP queue through which the agent receives messages from a sender.
-data StoredRcvQueue (q :: QueueStored) = RcvQueue
+data StoredRcvQueue (q :: DBStored) = RcvQueue
   { userId :: UserId,
     connId :: ConnId,
     server :: SMPServer,
@@ -98,7 +87,7 @@ data StoredRcvQueue (q :: QueueStored) = RcvQueue
     -- | queue status
     status :: QueueStatus,
     -- | database queue ID (within connection)
-    dbQueueId :: DBQueueId q,
+    dbQueueId :: DBEntityId' q,
     -- | True for a primary or a next primary queue of the connection (next if dbReplaceQueueId is set)
     primary :: Bool,
     -- | database queue ID to replace, Nothing if this queue is not replacing another, `Just Nothing` is used for replacing old queues
@@ -160,12 +149,12 @@ data InvShortLink = InvShortLink
   }
   deriving (Show)
 
-type SndQueue = StoredSndQueue 'QSStored
+type SndQueue = StoredSndQueue 'DBStored
 
-type NewSndQueue = StoredSndQueue 'QSNew
+type NewSndQueue = StoredSndQueue 'DBNew
 
 -- | A send queue. SMP queue through which the agent sends messages to a recipient.
-data StoredSndQueue (q :: QueueStored) = SndQueue
+data StoredSndQueue (q :: DBStored) = SndQueue
   { userId :: UserId,
     connId :: ConnId,
     server :: SMPServer,
@@ -184,7 +173,7 @@ data StoredSndQueue (q :: QueueStored) = SndQueue
     -- | queue status
     status :: QueueStatus,
     -- | database queue ID (within connection)
-    dbQueueId :: DBQueueId q,
+    dbQueueId :: DBEntityId' q,
     -- | True for a primary or a next primary queue of the connection (next if dbReplaceQueueId is set)
     primary :: Bool,
     -- | ID of the queue this one is replacing
@@ -257,7 +246,7 @@ instance SMPQueueRec RcvQueue where
   {-# INLINE qUserId #-}
   qConnId RcvQueue {connId} = connId
   {-# INLINE qConnId #-}
-  dbQId RcvQueue {dbQueueId = DBQueueId qId} = qId
+  dbQId RcvQueue {dbQueueId = DBEntityId qId} = qId
   {-# INLINE dbQId #-}
   dbReplaceQId RcvQueue {dbReplaceQueueId} = dbReplaceQueueId
   {-# INLINE dbReplaceQId #-}
@@ -267,7 +256,7 @@ instance SMPQueueRec SndQueue where
   {-# INLINE qUserId #-}
   qConnId SndQueue {connId} = connId
   {-# INLINE qConnId #-}
-  dbQId SndQueue {dbQueueId = DBQueueId qId} = qId
+  dbQId SndQueue {dbQueueId = DBEntityId qId} = qId
   {-# INLINE dbQId #-}
   dbReplaceQId SndQueue {dbReplaceQueueId} = dbReplaceQueueId
   {-# INLINE dbReplaceQId #-}

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

@@ -283,6 +283,7 @@ import Simplex.Messaging.Notifications.Types
 import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Protocol
 import qualified Simplex.Messaging.Protocol as SMP
+import Simplex.Messaging.Agent.Store.Entity
 import Simplex.Messaging.Transport.Client (TransportHost)
 import Simplex.Messaging.Util (bshow, catchAllErrors, eitherToMaybe, firstRow, firstRow', ifM, maybeFirstRow, tshow, ($>>=), (<$$>))
 import Simplex.Messaging.Version.Internal
@@ -858,7 +859,7 @@ createRcvMsg db connId rq@RcvQueue {dbQueueId} rcvMsgData@RcvMsgData {msgMeta =
   updateRcvMsgHash db connId sndMsgId internalRcvId internalHash
   setLastBrokerTs db connId dbQueueId brokerTs
 
-setLastBrokerTs :: DB.Connection -> ConnId -> DBQueueId 'QSStored -> UTCTime -> IO ()
+setLastBrokerTs :: DB.Connection -> ConnId -> DBEntityId -> 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)
 
@@ -1212,7 +1213,7 @@ getSndRatchet db connId v =
     DB.query db "SELECT ratchet_state, x3dh_pub_key_1, x3dh_pub_key_2, pq_pub_kem FROM ratchets WHERE conn_id = ?" (Only connId)
   where
     result = \case
-      (Just ratchetState, Just k1, Just k2, pKem_) -> 
+      (Just ratchetState, Just k1, Just k2, pKem_) ->
         let params = case pKem_ of
               Nothing -> CR.AE2ERatchetParams CR.SRKSProposed (CR.E2ERatchetParams v k1 k2 Nothing)
               Just (CR.ARKP s pKem) -> CR.AE2ERatchetParams s (CR.E2ERatchetParams v k1 k2 (Just pKem))
@@ -1811,15 +1812,6 @@ instance ToField QueueStatus where toField = toField . serializeQueueStatus
 
 instance FromField QueueStatus where fromField = fromTextField_ queueStatusT
 
-instance ToField (DBQueueId 'QSStored) where toField (DBQueueId qId) = toField qId
-
-instance FromField (DBQueueId 'QSStored) where 
-#if defined(dbPostgres)
-  fromField x dat = DBQueueId <$> fromField x dat
-#else
-  fromField x = DBQueueId <$> fromField x
-#endif
-
 instance ToField InternalRcvId where toField (InternalRcvId x) = toField x
 
 deriving newtype instance FromField InternalRcvId
@@ -2018,13 +2010,13 @@ insertSndQueue_ db connId' sq@SndQueue {..} serverKeyHash_ = do
         smp_client_version=EXCLUDED.smp_client_version,
         server_key_hash=EXCLUDED.server_key_hash
     |]
-    ((host server, port server, sndId, queueMode, connId', sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret) 
+    ((host server, port server, sndId, queueMode, connId', sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret)
     :. (status, qId, BI primary, dbReplaceQueueId, smpClientVersion, serverKeyHash_))
   pure (sq :: NewSndQueue) {connId = connId', dbQueueId = qId}
 
-newQueueId_ :: [Only Int64] -> DBQueueId 'QSStored
-newQueueId_ [] = DBQueueId 1
-newQueueId_ (Only maxId : _) = DBQueueId (maxId + 1)
+newQueueId_ :: [Only Int64] -> DBEntityId
+newQueueId_ [] = DBEntityId 1
+newQueueId_ (Only maxId : _) = DBEntityId (maxId + 1)
 
 -- * getConn helpers
 
@@ -2160,7 +2152,7 @@ rcvQueueQuery =
 
 toRcvQueue ::
   (UserId, C.KeyHash, ConnId, NonEmpty TransportHost, ServiceName, SMP.RecipientId, SMP.RcvPrivateAuthKey, SMP.RcvDhSecret, C.PrivateKeyX25519, Maybe C.DhSecretX25519, SMP.SenderId, Maybe QueueMode)
-    :. (QueueStatus, DBQueueId 'QSStored, BoolInt, Maybe Int64, Maybe RcvSwitchStatus, Maybe VersionSMPC, Int)
+    :. (QueueStatus, DBEntityId, BoolInt, Maybe Int64, Maybe RcvSwitchStatus, Maybe VersionSMPC, Int)
     :. (Maybe SMP.NtfPublicAuthKey, Maybe SMP.NtfPrivateAuthKey, Maybe SMP.NotifierId, Maybe RcvNtfDhSecret)
     :. (Maybe SMP.LinkId, Maybe LinkKey, Maybe C.PrivateKeyEd25519, Maybe EncDataBytes) ->
   RcvQueue
@@ -2210,7 +2202,7 @@ sndQueueQuery =
 toSndQueue ::
   (UserId, C.KeyHash, ConnId, NonEmpty TransportHost, ServiceName, SenderId, Maybe QueueMode)
     :. (Maybe SndPublicAuthKey, SndPrivateAuthKey, Maybe C.PublicKeyX25519, C.DhSecretX25519, QueueStatus)
-    :. (DBQueueId 'QSStored, BoolInt, Maybe Int64, Maybe SndSwitchStatus, VersionSMPC) ->
+    :. (DBEntityId, BoolInt, Maybe Int64, Maybe SndSwitchStatus, VersionSMPC) ->
   SndQueue
 toSndQueue
   ( (userId, keyHash, connId, host, port, sndId, queueMode)

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

@@ -0,0 +1,72 @@
+{-# LANGUAGE CPP #-}
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE KindSignatures #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE StandaloneDeriving #-}
+{-# LANGUAGE TypeApplications #-}
+
+module Simplex.Messaging.Agent.Store.Entity where
+
+import Data.Aeson (FromJSON (..), ToJSON (..))
+import qualified Data.Aeson as J
+import qualified Data.Aeson.Encoding as JE
+import Data.Int (Int64)
+import Data.Scientific (floatingOrInteger)
+import Simplex.Messaging.Agent.Store.DB (FromField (..), ToField (..))
+
+data DBStored = DBStored | DBNew
+
+data SDBStored (s :: DBStored) where
+  SDBStored :: SDBStored 'DBStored
+  SDBNew :: SDBStored 'DBNew
+
+deriving instance Show (SDBStored s)
+
+class DBStoredI s where sdbStored :: SDBStored s
+
+instance DBStoredI 'DBStored where sdbStored = SDBStored
+
+instance DBStoredI 'DBNew where sdbStored = SDBNew
+
+data DBEntityId' (s :: DBStored) where
+  DBEntityId :: Int64 -> DBEntityId' 'DBStored
+  DBNewEntity :: DBEntityId' 'DBNew
+
+deriving instance Show (DBEntityId' s)
+
+deriving instance Eq (DBEntityId' s)
+
+type DBEntityId = DBEntityId' 'DBStored
+
+type DBNewEntity = DBEntityId' 'DBNew
+
+instance ToJSON (DBEntityId' s) where
+  toEncoding = \case
+    DBEntityId i -> toEncoding i
+    DBNewEntity -> JE.null_
+  toJSON = \case
+    DBEntityId i -> toJSON i
+    DBNewEntity -> J.Null
+
+instance DBStoredI s => FromJSON (DBEntityId' s) where
+  parseJSON v = case (v, sdbStored @s) of
+    (J.Null, SDBNew) -> pure DBNewEntity
+    (J.Number n, SDBStored) -> case floatingOrInteger n of
+      Left (_ :: Double) -> fail "bad DBEntityId"
+      Right i -> pure $ DBEntityId (fromInteger i)
+    _ -> fail "bad DBEntityId"
+  omittedField = case sdbStored @s of
+    SDBStored -> Nothing
+    SDBNew -> Just DBNewEntity
+
+instance FromField DBEntityId where
+#if defined(dbPostgres)
+  fromField x dat = DBEntityId <$> fromField x dat
+#else
+  fromField x = DBEntityId <$> fromField x
+#endif
+
+instance ToField DBEntityId where toField (DBEntityId i) = toField i

src/Simplex/Messaging/Client.hs

@@ -424,7 +424,7 @@ data ProtocolClientConfig v = ProtocolClientConfig
   { -- | size of TBQueue to use for server commands and responses
     qSize :: Natural,
     -- | default server port if port is not specified in ProtocolServer
-    defaultTransport :: (ServiceName, ATransport),
+    defaultTransport :: (ServiceName, ATransport 'TClient),
     -- | network configuration
     networkConfig :: NetworkConfig,
     clientALPN :: Maybe [ALPN],
@@ -553,7 +553,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
             msgQ
           }
 
-    runClient :: (ServiceName, ATransport) -> TransportHost -> PClient v err msg -> IO (Either (ProtocolClientError err) (ProtocolClient v err msg))
+    runClient :: (ServiceName, ATransport 'TClient) -> TransportHost -> PClient v err msg -> IO (Either (ProtocolClientError err) (ProtocolClient v err msg))
     runClient (port', ATransport t) useHost c = do
       cVar <- newEmptyTMVarIO
       let tcConfig = (transportClientConfig networkConfig useHost useSNI) {alpn = clientALPN}
@@ -567,7 +567,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
         Just (Left e) -> pure $ Left e
         Nothing -> killThread tId $> Left PCENetworkError
 
-    useTransport :: (ServiceName, ATransport)
+    useTransport :: (ServiceName, ATransport 'TClient)
     useTransport = case port srv of
       "" -> case protocolTypeI @(ProtoType msg) of
         SPSMP | smpWebPort -> ("443", transport @TLS)
@@ -581,7 +581,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
             _ -> 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 :: forall c. Transport c => TProxy c 'TClient -> PClient v err msg -> TMVar (Either (ProtocolClientError err) (ProtocolClient v err msg)) -> c 'TClient -> IO ()
     client _ c cVar h = do
       ks <- if agreeSecret then Just <$> atomically (C.generateKeyPair g) else pure Nothing
       runExceptT (protocolClientHandshake @v @err @msg h ks (keyHash srv) serverVRange proxyServer) >>= \case

src/Simplex/Messaging/Crypto.hs

@@ -179,8 +179,6 @@ module Simplex.Messaging.Crypto
     unPad,
 
     -- * X509 Certificates
-    SignedCertificate,
-    Certificate,
     signCertificate,
     signX509,
     verifyX509,
@@ -240,7 +238,7 @@ import Data.String
 import Data.Type.Equality
 import Data.Typeable (Proxy (Proxy), Typeable)
 import Data.Word (Word32)
-import Data.X509
+import qualified Data.X509 as X
 import Data.X509.Validation (Fingerprint (..), getFingerprint)
 import GHC.TypeLits (ErrorMessage (..), KnownNat, Nat, TypeError, natVal, type (+))
 import Network.Transport.Internal (decodeWord16, encodeWord16)
@@ -1160,12 +1158,12 @@ sign :: APrivateSignKey -> ByteString -> ASignature
 sign (APrivateSignKey a k) = ASignature a . sign' k
 {-# INLINE sign #-}
 
-signCertificate :: APrivateSignKey -> Certificate -> SignedCertificate
+signCertificate :: APrivateSignKey -> X.Certificate -> X.SignedCertificate
 signCertificate = signX509
 {-# INLINE signCertificate #-}
 
-signX509 :: (ASN1Object o, Eq o, Show o) => APrivateSignKey -> o -> SignedExact o
-signX509 key = fst . objectToSignedExact f
+signX509 :: (ASN1Object o, Eq o, Show o) => APrivateSignKey -> o -> X.SignedExact o
+signX509 key = fst . X.objectToSignedExact f
   where
     f bytes =
       ( signatureBytes $ sign key bytes,
@@ -1174,33 +1172,33 @@ signX509 key = fst . objectToSignedExact f
       )
 {-# INLINE signX509 #-}
 
-verifyX509 :: (ASN1Object o, Eq o, Show o) => APublicVerifyKey -> SignedExact o -> Either String o
+verifyX509 :: (ASN1Object o, Eq o, Show o) => APublicVerifyKey -> X.SignedExact o -> Either String o
 verifyX509 key exact = do
   signature <- case signedAlg of
-    SignatureALG_IntrinsicHash PubKeyALG_Ed25519 -> ASignature SEd25519 <$> decodeSignature signedSignature
-    SignatureALG_IntrinsicHash PubKeyALG_Ed448 -> ASignature SEd448 <$> decodeSignature signedSignature
+    X.SignatureALG_IntrinsicHash X.PubKeyALG_Ed25519 -> ASignature SEd25519 <$> decodeSignature signedSignature
+    X.SignatureALG_IntrinsicHash X.PubKeyALG_Ed448 -> ASignature SEd448 <$> decodeSignature signedSignature
     _ -> Left "unknown x509 signature algorithm"
-  if verify key signature $ getSignedData exact then Right signedObject else Left "bad signature"
+  if verify key signature $ X.getSignedData exact then Right signedObject else Left "bad signature"
   where
-    Signed {signedObject, signedAlg, signedSignature} = getSigned exact
+    X.Signed {signedObject, signedAlg, signedSignature} = X.getSigned exact
 {-# INLINE verifyX509 #-}
 
-certificateFingerprint :: SignedCertificate -> KeyHash
+certificateFingerprint :: X.SignedCertificate -> KeyHash
 certificateFingerprint = signedFingerprint
 {-# INLINE certificateFingerprint #-}
 
-signedFingerprint :: (ASN1Object o, Eq o, Show o) => SignedExact o -> KeyHash
+signedFingerprint :: (ASN1Object o, Eq o, Show o) => X.SignedExact o -> KeyHash
 signedFingerprint o = KeyHash fp
   where
-    Fingerprint fp = getFingerprint o HashSHA256
+    Fingerprint fp = getFingerprint o X.HashSHA256
 
 class SignatureAlgorithmX509 a where
-  signatureAlgorithmX509 :: a -> SignatureALG
+  signatureAlgorithmX509 :: a -> X.SignatureALG
 
 instance SignatureAlgorithm a => SignatureAlgorithmX509 (SAlgorithm a) where
   signatureAlgorithmX509 = \case
-    SEd25519 -> SignatureALG_IntrinsicHash PubKeyALG_Ed25519
-    SEd448 -> SignatureALG_IntrinsicHash PubKeyALG_Ed448
+    SEd25519 -> X.SignatureALG_IntrinsicHash X.PubKeyALG_Ed25519
+    SEd448 -> X.SignatureALG_IntrinsicHash X.PubKeyALG_Ed448
   {-# INLINE signatureAlgorithmX509 #-}
 
 instance SignatureAlgorithmX509 APrivateSignKey where
@@ -1217,31 +1215,31 @@ instance SignatureAlgorithmX509 pk => SignatureAlgorithmX509 (a, pk) where
   {-# INLINE signatureAlgorithmX509 #-}
 
 -- | A wrapper to marshall signed ASN1 objects, like certificates.
-newtype SignedObject a = SignedObject {getSignedExact :: SignedExact a}
+newtype SignedObject a = SignedObject {getSignedExact :: X.SignedExact a}
 
 instance (Typeable a, Eq a, Show a, ASN1Object a) => FromField (SignedObject a) where
 #if defined(dbPostgres)
-  fromField f dat = SignedObject <$> blobFieldDecoder decodeSignedObject f dat
+  fromField f dat = SignedObject <$> blobFieldDecoder X.decodeSignedObject f dat
 #else
-  fromField = fmap SignedObject . blobFieldDecoder decodeSignedObject
+  fromField = fmap SignedObject . blobFieldDecoder X.decodeSignedObject
 #endif
 
 instance (Eq a, Show a, ASN1Object a) => ToField (SignedObject a) where
-  toField (SignedObject s) = toField . Binary $ encodeSignedObject s
+  toField (SignedObject s) = toField . Binary $ X.encodeSignedObject s
 
 instance (Eq a, Show a, ASN1Object a) => Encoding (SignedObject a) where
-  smpEncode (SignedObject exact) = smpEncode . Large $ encodeSignedObject exact
-  smpP = fmap SignedObject . decodeSignedObject . unLarge <$?> smpP
+  smpEncode (SignedObject exact) = smpEncode . Large $ X.encodeSignedObject exact
+  smpP = fmap SignedObject . X.decodeSignedObject . unLarge <$?> smpP
 
-encodeCertChain :: CertificateChain -> L.NonEmpty Large
+encodeCertChain :: X.CertificateChain -> L.NonEmpty Large
 encodeCertChain cc = L.fromList $ map Large blobs
   where
-    CertificateChainRaw blobs = encodeCertificateChain cc
+    X.CertificateChainRaw blobs = X.encodeCertificateChain cc
 
-certChainP :: A.Parser CertificateChain
+certChainP :: A.Parser X.CertificateChain
 certChainP = do
-  rawChain <- CertificateChainRaw . map unLarge . L.toList <$> smpP
-  either (fail . show) pure $ decodeCertificateChain rawChain
+  rawChain <- X.CertificateChainRaw . map unLarge . L.toList <$> smpP
+  either (fail . show) pure $ X.decodeCertificateChain rawChain
 
 -- | Signature verification.
 --
@@ -1453,19 +1451,19 @@ xSalsa20 secret nonce msg = (rs, msg')
     (rs, state2) = XSalsa.generate state1 32
     (msg', _) = XSalsa.combine state2 msg
 
-publicToX509 :: PublicKey a -> PubKey
+publicToX509 :: PublicKey a -> X.PubKey
 publicToX509 = \case
-  PublicKeyEd25519 k -> PubKeyEd25519 k
-  PublicKeyEd448 k -> PubKeyEd448 k
-  PublicKeyX25519 k -> PubKeyX25519 k
-  PublicKeyX448 k -> PubKeyX448 k
+  PublicKeyEd25519 k -> X.PubKeyEd25519 k
+  PublicKeyEd448 k -> X.PubKeyEd448 k
+  PublicKeyX25519 k -> X.PubKeyX25519 k
+  PublicKeyX448 k -> X.PubKeyX448 k
 
-privateToX509 :: PrivateKey a -> PrivKey
+privateToX509 :: PrivateKey a -> X.PrivKey
 privateToX509 = \case
-  PrivateKeyEd25519 k _ -> PrivKeyEd25519 k
-  PrivateKeyEd448 k _ -> PrivKeyEd448 k
-  PrivateKeyX25519 k _ -> PrivKeyX25519 k
-  PrivateKeyX448 k _ -> PrivKeyX448 k
+  PrivateKeyEd25519 k _ -> X.PrivKeyEd25519 k
+  PrivateKeyEd448 k _ -> X.PrivKeyEd448 k
+  PrivateKeyX25519 k _ -> X.PrivKeyX25519 k
+  PrivateKeyX448 k _ -> X.PrivKeyX448 k
 
 encodeASNObj :: ASN1Object a => a -> ByteString
 encodeASNObj k = toStrict . encodeASN1 DER $ toASN1 k []
@@ -1478,20 +1476,20 @@ decodePubKey = decodeKey >=> x509ToPublic >=> pubKey
 decodePrivKey :: CryptoPrivateKey k => ByteString -> Either String k
 decodePrivKey = decodeKey >=> x509ToPrivate >=> privKey
 
-x509ToPublic :: (PubKey, [ASN1]) -> Either String APublicKey
+x509ToPublic :: (X.PubKey, [ASN1]) -> Either String APublicKey
 x509ToPublic = \case
-  (PubKeyEd25519 k, []) -> Right . APublicKey SEd25519 $ PublicKeyEd25519 k
-  (PubKeyEd448 k, []) -> Right . APublicKey SEd448 $ PublicKeyEd448 k
-  (PubKeyX25519 k, []) -> Right . APublicKey SX25519 $ PublicKeyX25519 k
-  (PubKeyX448 k, []) -> Right . APublicKey SX448 $ PublicKeyX448 k
+  (X.PubKeyEd25519 k, []) -> Right . APublicKey SEd25519 $ PublicKeyEd25519 k
+  (X.PubKeyEd448 k, []) -> Right . APublicKey SEd448 $ PublicKeyEd448 k
+  (X.PubKeyX25519 k, []) -> Right . APublicKey SX25519 $ PublicKeyX25519 k
+  (X.PubKeyX448 k, []) -> Right . APublicKey SX448 $ PublicKeyX448 k
   r -> keyError r
 
-x509ToPrivate :: (PrivKey, [ASN1]) -> Either String APrivateKey
+x509ToPrivate :: (X.PrivKey, [ASN1]) -> Either String APrivateKey
 x509ToPrivate = \case
-  (PrivKeyEd25519 k, []) -> Right . APrivateKey SEd25519 . PrivateKeyEd25519 k $ Ed25519.toPublic k
-  (PrivKeyEd448 k, []) -> Right . APrivateKey SEd448 . PrivateKeyEd448 k $ Ed448.toPublic k
-  (PrivKeyX25519 k, []) -> Right . APrivateKey SX25519 . PrivateKeyX25519 k $ X25519.toPublic k
-  (PrivKeyX448 k, []) -> Right . APrivateKey SX448 . PrivateKeyX448 k $ X448.toPublic k
+  (X.PrivKeyEd25519 k, []) -> Right . APrivateKey SEd25519 . PrivateKeyEd25519 k $ Ed25519.toPublic k
+  (X.PrivKeyEd448 k, []) -> Right . APrivateKey SEd448 . PrivateKeyEd448 k $ Ed448.toPublic k
+  (X.PrivKeyX25519 k, []) -> Right . APrivateKey SX25519 . PrivateKeyX25519 k $ X25519.toPublic k
+  (X.PrivKeyX448 k, []) -> Right . APrivateKey SX448 . PrivateKeyX448 k $ X448.toPublic k
   r -> keyError r
 
 decodeKey :: ASN1Object a => ByteString -> Either String (a, [ASN1])

src/Simplex/Messaging/Notifications/Server.hs

@@ -71,7 +71,7 @@ import Simplex.Messaging.Server.QueueStore (getSystemDate)
 import Simplex.Messaging.Server.Stats (PeriodStats (..), PeriodStatCounts (..), periodStatCounts, periodStatDataCounts, updatePeriodStats)
 import Simplex.Messaging.Session
 import Simplex.Messaging.TMap (TMap)
-import Simplex.Messaging.Transport (ATransport (..), THandle (..), THandleAuth (..), THandleParams (..), TProxy, Transport (..), TransportPeer (..), defaultSupportedParams)
+import Simplex.Messaging.Transport (ASrvTransport, ATransport (..), THandle (..), THandleAuth (..), THandleParams (..), TProxy, Transport (..), TransportPeer (..), defaultSupportedParams)
 import Simplex.Messaging.Transport.Buffer (trimCR)
 import Simplex.Messaging.Transport.Server (AddHTTP, runTransportServer, runLocalTCPServer)
 import Simplex.Messaging.Util
@@ -120,7 +120,7 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions}
     )
     `finally` stopServer
   where
-    runServer :: (ServiceName, ATransport, AddHTTP) -> M ()
+    runServer :: (ServiceName, ASrvTransport, AddHTTP) -> M ()
     runServer (tcpPort, ATransport t, _addHTTP) = do
       srvCreds <- asks tlsServerCreds
       serverSignKey <- either fail pure $ fromTLSCredentials srvCreds
@@ -128,7 +128,7 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions}
       liftIO $ runTransportServer started tcpPort defaultSupportedParams srvCreds (Just supportedNTFHandshakes) tCfg $ \h -> runClient serverSignKey t h `runReaderT` env
     fromTLSCredentials (_, pk) = C.x509ToPrivate (pk, []) >>= C.privKey
 
-    runClient :: Transport c => C.APrivateSignKey -> TProxy c -> c -> M ()
+    runClient :: Transport c => C.APrivateSignKey -> TProxy c 'TServer -> c 'TServer -> M ()
     runClient signKey _ h = do
       kh <- asks serverIdentity
       ks <- atomically . C.generateKeyPair =<< asks random

src/Simplex/Messaging/Notifications/Server/Env.hs

@@ -39,21 +39,20 @@ import Simplex.Messaging.Server.StoreLog (closeStoreLog)
 import Simplex.Messaging.Session
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Transport (ATransport, THandleParams, TransportPeer (..))
+import Simplex.Messaging.Transport (ASrvTransport, THandleParams, TransportPeer (..))
 import Simplex.Messaging.Transport.Server (AddHTTP, ServerCredentials, TransportServerConfig, loadFingerprint, loadServerCredential)
 import System.Exit (exitFailure)
 import System.Mem.Weak (Weak)
 import UnliftIO.STM
 
 data NtfServerConfig = NtfServerConfig
-  { transports :: [(ServiceName, ATransport, AddHTTP)],
+  { transports :: [(ServiceName, ASrvTransport, AddHTTP)],
     controlPort :: Maybe ServiceName,
     controlPortUserAuth :: Maybe BasicAuth,
     controlPortAdminAuth :: Maybe BasicAuth,
     subIdBytes :: Int,
     regCodeBytes :: Int,
     clientQSize :: Natural,
-    subQSize :: Natural,
     pushQSize :: Natural,
     smpAgentCfg :: SMPClientAgentConfig,
     apnsConfig :: APNSPushClientConfig,
@@ -94,11 +93,11 @@ data NtfEnv = NtfEnv
   }
 
 newNtfServerEnv :: NtfServerConfig -> IO NtfEnv
-newNtfServerEnv config@NtfServerConfig {subQSize, pushQSize, smpAgentCfg, apnsConfig, dbStoreConfig, ntfCredentials, startOptions} = do
+newNtfServerEnv config@NtfServerConfig {pushQSize, smpAgentCfg, apnsConfig, dbStoreConfig, ntfCredentials, startOptions} = do
   when (compactLog startOptions) $ compactDbStoreLog $ dbStoreLogPath dbStoreConfig
   random <- C.newRandom
   store <- newNtfDbStore dbStoreConfig
-  subscriber <- newNtfSubscriber subQSize smpAgentCfg random
+  subscriber <- newNtfSubscriber smpAgentCfg random
   pushServer <- newNtfPushServer pushQSize apnsConfig
   tlsServerCreds <- loadServerCredential ntfCredentials
   Fingerprint fp <- loadFingerprint ntfCredentials
@@ -121,8 +120,8 @@ data NtfSubscriber = NtfSubscriber
 
 type SMPSubscriberVar = SessionVar SMPSubscriber
 
-newNtfSubscriber :: Natural -> SMPClientAgentConfig -> TVar ChaChaDRG -> IO NtfSubscriber
-newNtfSubscriber qSize smpAgentCfg random = do
+newNtfSubscriber :: SMPClientAgentConfig -> TVar ChaChaDRG -> IO NtfSubscriber
+newNtfSubscriber smpAgentCfg random = do
   smpSubscribers <- TM.emptyIO
   subscriberSeq <- newTVarIO 0
   smpAgent <- newSMPClientAgent smpAgentCfg random

src/Simplex/Messaging/Notifications/Server/Main.hs

@@ -46,7 +46,7 @@ import Simplex.Messaging.Server.Main (strParse)
 import Simplex.Messaging.Server.Main.Init (iniDbOpts)
 import Simplex.Messaging.Server.QueueStore.Postgres.Config (PostgresStoreCfg (..))
 import Simplex.Messaging.Server.StoreLog (closeStoreLog)
-import Simplex.Messaging.Transport (ATransport, simplexMQVersion)
+import Simplex.Messaging.Transport (ASrvTransport, simplexMQVersion)
 import Simplex.Messaging.Transport.Client (TransportHost (..))
 import Simplex.Messaging.Transport.Server (AddHTTP, ServerCredentials (..), TransportServerConfig (..), defaultTransportServerConfig)
 import Simplex.Messaging.Util (eitherToMaybe, ifM, tshow)
@@ -233,7 +233,6 @@ ntfServerCLI cfgPath logPath =
               subIdBytes = 24,
               regCodeBytes = 32,
               clientQSize = 64,
-              subQSize = 2048,
               pushQSize = 32768,
               smpAgentCfg =
                 defaultSMPClientAgentConfig
@@ -287,7 +286,7 @@ ntfServerCLI cfgPath logPath =
       putStrLn "Configure notification server storage."
       exitFailure
 
-printNtfServerConfig :: [(ServiceName, ATransport, AddHTTP)] -> PostgresStoreCfg -> IO ()
+printNtfServerConfig :: [(ServiceName, ASrvTransport, AddHTTP)] -> PostgresStoreCfg -> IO ()
 printNtfServerConfig transports PostgresStoreCfg {dbOpts = DBOpts {connstr, schema}, dbStoreLogPath} = do
   B.putStrLn $ "PostgreSQL database: " <> connstr <> ", schema: " <> schema
   printServerConfig "NTF" transports dbStoreLogPath

src/Simplex/Messaging/Notifications/Transport.hs

@@ -110,7 +110,7 @@ instance Encoding NtfClientHandshake where
     pure NtfClientHandshake {ntfVersion, keyHash}
 
 -- | Notifcations server transport handshake.
-ntfServerHandshake :: forall c. Transport c => C.APrivateSignKey -> c -> C.KeyPairX25519 -> C.KeyHash -> VersionRangeNTF -> ExceptT TransportError IO (THandleNTF c 'TServer)
+ntfServerHandshake :: forall c. Transport c => C.APrivateSignKey -> c 'TServer -> C.KeyPairX25519 -> C.KeyHash -> VersionRangeNTF -> ExceptT TransportError IO (THandleNTF c 'TServer)
 ntfServerHandshake serverSignKey c (k, pk) kh ntfVRange = do
   let th@THandle {params = THandleParams {sessionId}} = ntfTHandle c
   let sk = C.signX509 serverSignKey $ C.publicToX509 k
@@ -126,7 +126,7 @@ ntfServerHandshake serverSignKey c (k, pk) kh ntfVRange = do
             Nothing -> throwE TEVersion
 
 -- | Notifcations server client transport handshake.
-ntfClientHandshake :: forall c. Transport c => c -> C.KeyHash -> VersionRangeNTF -> Bool -> ExceptT TransportError IO (THandleNTF c 'TClient)
+ntfClientHandshake :: forall c. Transport c => c 'TClient -> C.KeyHash -> VersionRangeNTF -> Bool -> ExceptT TransportError IO (THandleNTF c 'TClient)
 ntfClientHandshake c keyHash ntfVRange _proxyServer = do
   let th@THandle {params = THandleParams {sessionId}} = ntfTHandle c
   NtfServerHandshake {sessionId = sessId, ntfVersionRange, authPubKey = sk'} <- getHandshake th
@@ -137,7 +137,7 @@ ntfClientHandshake c keyHash ntfVRange _proxyServer = do
         ck_ <- forM sk' $ \signedKey -> liftEitherWith (const $ TEHandshake BAD_AUTH) $ do
           serverKey <- getServerVerifyKey c
           pubKey <- C.verifyX509 serverKey signedKey
-          (,(getServerCerts c, signedKey)) <$> (C.x509ToPublic (pubKey, []) >>= C.pubKey)
+          (,(getPeerCertChain c, signedKey)) <$> (C.x509ToPublic (pubKey, []) >>= C.pubKey)
         let v = maxVersion vr
         sendHandshake th $ NtfClientHandshake {ntfVersion = v, keyHash}
         pure $ ntfThHandleClient th v vr ck_
@@ -160,7 +160,7 @@ ntfThHandle_ th@THandle {params} v vr thAuth =
       params' = params {thVersion = v, thServerVRange = vr, thAuth, implySessId = v3, batch = v3}
    in (th :: THandleNTF c p) {params = params'}
 
-ntfTHandle :: Transport c => c -> THandleNTF c p
+ntfTHandle :: Transport c => c p -> THandleNTF c p
 ntfTHandle c = THandle {connection = c, params}
   where
     v = VersionNTF 0

src/Simplex/Messaging/Protocol.hs

@@ -305,7 +305,7 @@ data SParty :: Party -> Type where
   SRecipient :: SParty Recipient
   SSender :: SParty Sender
   SNotifier :: SParty Notifier
-  SSenderLink :: SParty LinkClient 
+  SSenderLink :: SParty LinkClient
   SProxiedClient :: SParty ProxiedClient
 
 instance TestEquality SParty where
@@ -1466,7 +1466,7 @@ transmissionP THandleParams {sessionId, implySessId} = do
 class (ProtocolTypeI (ProtoType msg), ProtocolEncoding v err msg, ProtocolEncoding v err (ProtoCommand msg), Show err, Show msg) => Protocol v err msg | msg -> v, msg -> err where
   type ProtoCommand msg = cmd | cmd -> msg
   type ProtoType msg = (sch :: ProtocolType) | sch -> msg
-  protocolClientHandshake :: forall c. Transport c => c -> Maybe C.KeyPairX25519 -> C.KeyHash -> VersionRange v -> Bool -> ExceptT TransportError IO (THandle v c 'TClient)
+  protocolClientHandshake :: forall c. Transport c => c 'TClient -> Maybe C.KeyPairX25519 -> C.KeyHash -> VersionRange v -> Bool -> ExceptT TransportError IO (THandle v c 'TClient)
   protocolPing :: ProtoCommand msg
   protocolError :: msg -> Maybe err
 

src/Simplex/Messaging/Server.hs

@@ -79,6 +79,7 @@ import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Data.Time.Format.ISO8601 (iso8601Show)
 import Data.Type.Equality
 import Data.Typeable (cast)
+import qualified Data.X509 as X
 import GHC.Conc.Signal
 import GHC.IORef (atomicSwapIORef)
 import GHC.Stats (getRTSStats)
@@ -87,7 +88,7 @@ import Network.Socket (ServiceName, Socket, socketToHandle)
 import qualified Network.TLS as TLS
 import Numeric.Natural (Natural)
 import Simplex.Messaging.Agent.Lock
-import Simplex.Messaging.Client (ProtocolClient (thParams), ProtocolClientError (..), SMPClient, SMPClientError, forwardSMPTransmission, smpProxyError, temporaryClientError)
+import Simplex.Messaging.Client (ProtocolClient (thParams), ProtocolClientError (..), SMPClient, SMPClientError, forwardSMPTransmission, nonBlockingWriteTBQueue, smpProxyError, temporaryClientError)
 import Simplex.Messaging.Client.Agent (OwnServer, SMPClientAgent (..), SMPClientAgentEvent (..), closeSMPClientAgent, getSMPServerClient'', isOwnServer, lookupSMPServerClient, getConnectedSMPServerClient)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
@@ -162,8 +163,8 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
     stopServer s
     liftIO $ exitSuccess
   raceAny_
-    ( serverThread s "server subscribedQ" subscribedQ subscribers subClients pendingSubEvents subscriptions cancelSub
-        : serverThread s "server ntfSubscribedQ" ntfSubscribedQ Env.notifiers ntfSubClients pendingNtfSubEvents ntfSubscriptions (\_ -> pure ())
+    ( serverThread "server subscribers" s subscribers subscriptions cancelSub
+        : serverThread "server ntfSubscribers" s ntfSubscribers ntfSubscriptions (\_ -> pure ())
         : deliverNtfsThread s
         : sendPendingEvtsThread s
         : receiveFromProxyAgent pa
@@ -177,28 +178,28 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
     )
     `finally` stopServer s
   where
-    runServer :: (ServiceName, ATransport, AddHTTP) -> M ()
+    runServer :: (ServiceName, ASrvTransport, AddHTTP) -> M ()
     runServer (tcpPort, ATransport t, addHTTP) = do
-      smpCreds <- asks tlsServerCreds
+      smpCreds@(srvCert, srvKey) <- asks tlsServerCreds
       httpCreds_ <- asks httpServerCreds
       ss <- liftIO newSocketState
       asks sockets >>= atomically . (`modifyTVar'` ((tcpPort, ss) :))
-      serverSignKey <- either fail pure $ fromTLSCredentials smpCreds
+      srvSignKey <- either fail pure $ fromTLSPrivKey srvKey
       env <- ask
       liftIO $ case (httpCreds_, attachHTTP_) of
         (Just httpCreds, Just attachHTTP) | addHTTP ->
           runTransportServerState_ ss started tcpPort defaultSupportedParamsHTTPS chooseCreds (Just combinedALPNs) tCfg $ \s h ->
             case cast h of
-              Just TLS {tlsContext} | maybe False (`elem` httpALPN) (getSessionALPN h) -> labelMyThread "https client" >> attachHTTP s tlsContext
-              _ -> runClient serverSignKey t h `runReaderT` env
+              Just (TLS {tlsContext} :: TLS 'TServer) | maybe False (`elem` httpALPN) (getSessionALPN h) -> labelMyThread "https client" >> attachHTTP s tlsContext
+              _ -> runClient srvCert srvSignKey t h `runReaderT` env
           where
             chooseCreds = maybe smpCreds (\_host -> httpCreds)
             combinedALPNs = supportedSMPHandshakes <> httpALPN
             httpALPN :: [ALPN]
             httpALPN = ["h2", "http/1.1"]
         _ ->
-          runTransportServerState ss started tcpPort defaultSupportedParams smpCreds (Just supportedSMPHandshakes) tCfg $ \h -> runClient serverSignKey t h `runReaderT` env
-    fromTLSCredentials (_, pk) = C.x509ToPrivate (pk, []) >>= C.privKey
+          runTransportServerState ss started tcpPort defaultSupportedParams smpCreds (Just supportedSMPHandshakes) tCfg $ \h -> runClient srvCert srvSignKey t h `runReaderT` env
+    fromTLSPrivKey pk = C.x509ToPrivate (pk, []) >>= C.privKey
 
     sigIntHandlerThread :: M ()
     sigIntHandlerThread = do
@@ -229,66 +230,63 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
 
     serverThread ::
       forall s.
-      Server ->
       String ->
-      (Server -> TQueue (QueueId, ClientId, Subscribed)) ->
-      (Server -> TMap QueueId (TVar AClient)) ->
-      (Server -> TVar (IM.IntMap AClient)) ->
-      (Server -> TVar (IM.IntMap (NonEmpty (QueueId, Subscribed)))) ->
+      Server ->
+      (Server -> ServerSubscribers) ->
       (forall st. Client st -> TMap QueueId s) ->
       (s -> IO ()) ->
       M ()
-    serverThread s label subQ subs subClnts pendingEvts clientSubs unsub = do
+    serverThread label srv srvSubscribers clientSubs unsub = do
       labelMyThread label
-      cls <- asks clients
-      liftIO . forever $
-        (atomically (readTQueue $ subQ s) >>= atomically . updateSubscribers cls)
+      liftIO . forever $ do
+        -- Reading clients outside of `updateSubscribers` transaction to avoid transaction re-evaluation on each new connected client.
+        -- In case client disconnects during the transaction (its `connected` property is read),
+        -- the transaction will still be re-evaluated, and the client won't be stored as subscribed.
+        sub@(_, clntId, _) <- atomically $ readTQueue subQ
+        c_ <- getServerClient clntId srv
+        atomically (updateSubscribers c_ sub)
           $>>= endPreviousSubscriptions
           >>= mapM_ unsub
       where
-        updateSubscribers :: TVar (IM.IntMap (Maybe AClient)) -> (QueueId, ClientId, Subscribed) -> STM (Maybe ((QueueId, Subscribed), AClient))
-        updateSubscribers cls (qId, clntId, subscribed) =
-          -- Client lookup by ID is in the same STM transaction.
-          -- In case client disconnects during the transaction,
-          -- it will be re-evaluated, and the client won't be stored as subscribed.
-          (readTVar cls >>= updateSub . IM.lookup clntId)
-            $>>= clientToBeNotified
+        ServerSubscribers {subQ, queueSubscribers, subClients, pendingEvents} = srvSubscribers srv
+        updateSubscribers :: Maybe AClient -> (QueueId, ClientId, Subscribed) -> STM (Maybe ((QueueId, BrokerMsg), AClient))
+        updateSubscribers c_ (qId, clntId, subscribed) = updateSub $>>= clientToBeNotified
           where
-            ss = subs s
-            updateSub = \case
-              Just (Just clnt)
-                | subscribed -> do
-                    modifyTVar' (subClnts s) $ IM.insert clntId clnt -- add client to server's subscribed cients
-                    TM.lookup qId ss >>= -- insert subscribed and current client
-                      maybe
-                        (newTVar clnt >>= \cv -> TM.insert qId cv ss $> Nothing)
-                        (\cv -> Just <$> swapTVar cv clnt)
-                | otherwise -> do
-                    removeWhenNoSubs clnt
-                    TM.lookupDelete qId ss >>= mapM readTVar
-              -- This case catches Just Nothing - it cannot happen here.
-              -- Nothing is there only before client thread is started.
-              _ -> TM.lookup qId ss >>= mapM readTVar -- do not insert client if it is already disconnected, but send END to any other client
-            clientToBeNotified ac@(AClient _ _ c')
-              | clntId == clientId c' = pure Nothing
-              | otherwise = (\yes -> if yes then Just ((qId, subscribed), ac) else Nothing) <$> readTVar (connected c')
-        endPreviousSubscriptions :: ((QueueId, Subscribed), AClient) -> IO (Maybe s)
-        endPreviousSubscriptions (qEvt@(qId, _), ac@(AClient _ _ c)) = do
-          atomically $ modifyTVar' (pendingEvts s) $ IM.alter (Just . maybe [qEvt] (qEvt <|)) (clientId c)
+            updateSub = case c_ of
+              Just c@(AClient _ _ Client {connected}) -> ifM (readTVar connected) (updateSubConnected c) updateSubDisconnected
+              Nothing -> updateSubDisconnected
+            updateSubConnected c
+              | subscribed = do
+                  modifyTVar' subClients $ IS.insert clntId -- add client to server's subscribed cients
+                  upsertSubscribedClient qId c queueSubscribers
+              | otherwise = do
+                  removeWhenNoSubs c
+                  lookupDeleteSubscribedClient qId queueSubscribers
+            -- do not insert client if it is already disconnected, but send END to any other client
+            updateSubDisconnected = lookupDeleteSubscribedClient qId queueSubscribers
+            clientToBeNotified ac@(AClient _ _ Client {clientId, connected})
+              | clntId == clientId = pure Nothing
+              | otherwise = (\yes -> if yes then Just ((qId, subEvt), ac) else Nothing) <$> readTVar connected
+              where
+                subEvt = if subscribed then END else DELD
+        endPreviousSubscriptions :: ((QueueId, BrokerMsg), AClient) -> IO (Maybe s)
+        endPreviousSubscriptions (evt@(qId, _), ac@(AClient _ _ c)) = do
+          atomically $ modifyTVar' pendingEvents $ IM.alter (Just . maybe [evt] (evt <|)) (clientId c)
           atomically $ do
             sub <- TM.lookupDelete qId (clientSubs c)
             removeWhenNoSubs ac $> sub
         -- remove client from server's subscribed cients
-        removeWhenNoSubs (AClient _ _ c) = whenM (null <$> readTVar (clientSubs c)) $ modifyTVar' (subClnts s) $ IM.delete (clientId c)
+        removeWhenNoSubs (AClient _ _ c) = whenM (null <$> readTVar (clientSubs c)) $ modifyTVar' subClients $ IS.delete (clientId c)
 
     deliverNtfsThread :: Server -> M ()
-    deliverNtfsThread Server {ntfSubClients} = do
+    deliverNtfsThread srv@Server {ntfSubscribers} = do
       ntfInt <- asks $ ntfDeliveryInterval . config
       NtfStore ns <- asks ntfStore
       stats <- asks serverStats
       liftIO $ forever $ do
         threadDelay ntfInt
-        readTVarIO ntfSubClients >>= mapM_ (deliverNtfs ns stats)
+        cIds <- IS.toList <$> readTVarIO (subClients ntfSubscribers)
+        forM_ cIds $ \cId -> getServerClient cId srv >>= mapM_ (deliverNtfs ns stats)
       where
         deliverNtfs ns stats (AClient _ _ Client {clientId, ntfSubscriptions, sndQ, connected}) =
           whenM (currentClient readTVarIO) $ do
@@ -308,7 +306,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
                 writeTBQueue sndQ ts
               pure $ length ts_
             currentClient :: Monad m => (forall a. TVar a -> m a) -> m Bool
-            currentClient rd = (&&) <$> rd connected <*> (IM.member clientId <$> rd ntfSubClients)
+            currentClient rd = (&&) <$> rd connected <*> (IS.member clientId <$> rd (subClients ntfSubscribers))
             addNtfs :: [Transmission BrokerMsg] -> (NotifierId, TVar [MsgNtf]) -> STM [Transmission BrokerMsg]
             addNtfs acc (nId, v) =
               readTVar v >>= \case
@@ -324,37 +322,30 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
               atomicModifyIORef'_ (msgNtfsB stats) (+ (len `div` 80 + 1)) -- up to 80 NMSG in the batch
 
     sendPendingEvtsThread :: Server -> M ()
-    sendPendingEvtsThread s = do
+    sendPendingEvtsThread srv@Server {subscribers, ntfSubscribers} = do
       endInt <- asks $ pendingENDInterval . config
-      cls <- asks clients
-      forever $ do
+      stats <- asks serverStats
+      liftIO $ forever $ do
         threadDelay endInt
-        sendPending cls $ pendingSubEvents s
-        sendPending cls $ pendingNtfSubEvents s
+        sendPending subscribers stats
+        sendPending ntfSubscribers stats
       where
-        sendPending cls ref = do
-          ends <- atomically $ swapTVar ref IM.empty
-          unless (null ends) $ forM_ (IM.assocs ends) $ \(cId, qEvts) ->
-            mapM_ (queueEvts qEvts) . join . IM.lookup cId =<< readTVarIO cls
-        queueEvts qEvts (AClient _ _ c@Client {connected, sndQ = q}) =
-          whenM (readTVarIO connected) $ do
-            sent <- atomically $ tryWriteTBQueue q ts
-            if sent
-              then updateEndStats
-              else -- if queue is full it can block
-                forkClient c ("sendPendingEvtsThread.queueEvts") $
-                  atomically (writeTBQueue q ts) >> updateEndStats
+        sendPending ServerSubscribers {pendingEvents} stats = do
+          pending <- atomically $ swapTVar pendingEvents IM.empty
+          unless (null pending) $ forM_ (IM.assocs pending) $ \(cId, evts) ->
+            getServerClient cId srv >>= mapM_ (enqueueEvts evts)
           where
-            ts = L.map (\(qId, subscribed) -> (CorrId "", qId, evt subscribed)) qEvts
-            evt True = END
-            evt False = DELD
-            -- this accounts for both END and DELD events
-            updateEndStats = do
-              stats <- asks serverStats
-              let len = L.length qEvts
-              when (len > 0) $ liftIO $ do
-                atomicModifyIORef'_ (qSubEnd stats) (+ len)
-                atomicModifyIORef'_ (qSubEndB stats) (+ (len `div` 255 + 1)) -- up to 255 ENDs or DELDs in the batch
+            enqueueEvts evts (AClient _ _ Client {connected, sndQ}) =
+              whenM (readTVarIO connected) $
+                nonBlockingWriteTBQueue sndQ ts >> updateEndStats
+              where
+                ts = L.map (\(qId, evt) -> (CorrId "", qId, evt)) evts
+                -- this accounts for both END and DELD events
+                updateEndStats = do
+                  let len = L.length evts
+                  when (len > 0) $ do
+                    atomicModifyIORef'_ (qSubEnd stats) (+ len)
+                    atomicModifyIORef'_ (qSubEndB stats) (+ (len `div` 255 + 1)) -- up to 255 ENDs or DELDs in the batch
 
     receiveFromProxyAgent :: ProxyAgent -> M ()
     receiveFromProxyAgent ProxyAgent {smpAgent = SMPClientAgent {agentQ}} =
@@ -581,28 +572,31 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
       pure ServerMetrics {statsData = d, activeQueueCounts = ps, activeNtfCounts = psNtf, queueCount, notifierCount, rtsOptions}
 
     getRealTimeMetrics :: Env -> IO RealTimeMetrics
-    getRealTimeMetrics Env {clients, sockets, msgStore = AMS _ _ ms, server = Server {subscribers, notifiers, subClients, ntfSubClients}} = do
+    getRealTimeMetrics Env {sockets, msgStore = AMS _ _ ms, server = srv@Server {subscribers, ntfSubscribers}} = do
       socketStats <- mapM (traverse getSocketStats) =<< readTVarIO sockets
 #if MIN_VERSION_base(4,18,0)
       threadsCount <- length <$> listThreads
 #else
       let threadsCount = 0
 #endif
-      clientsCount <- IM.size <$> readTVarIO clients
-      smpSubsCount <- M.size <$> readTVarIO subscribers
-      smpSubClientsCount <- IM.size <$> readTVarIO subClients
-      ntfSubsCount <- M.size <$> readTVarIO notifiers
-      ntfSubClientsCount <- IM.size <$> readTVarIO ntfSubClients
+      clientsCount <- IM.size <$> getServerClients srv
+      smpSubs <- getSubscribersMetrics subscribers
+      ntfSubs <- getSubscribersMetrics ntfSubscribers
       loadedCounts <- loadedQueueCounts ms
-      pure RealTimeMetrics {socketStats, threadsCount, clientsCount, smpSubsCount, smpSubClientsCount, ntfSubsCount, ntfSubClientsCount, loadedCounts}
+      pure RealTimeMetrics {socketStats, threadsCount, clientsCount, smpSubs, ntfSubs, loadedCounts}
+      where
+        getSubscribersMetrics ServerSubscribers {queueSubscribers, subClients} = do
+          subsCount <- M.size <$> getSubscribedClients queueSubscribers
+          subClientsCount <- IS.size <$> readTVarIO subClients
+          pure RTSubscriberMetrics {subsCount, subClientsCount}
 
-    runClient :: Transport c => C.APrivateSignKey -> TProxy c -> c -> M ()
-    runClient signKey tp h = do
+    runClient :: Transport c => X.CertificateChain -> C.APrivateSignKey -> TProxy c 'TServer -> c 'TServer -> M ()
+    runClient srvCert srvSignKey tp h = do
       kh <- asks serverIdentity
       ks <- atomically . C.generateKeyPair =<< asks random
       ServerConfig {smpServerVRange, smpHandshakeTimeout} <- asks config
       labelMyThread $ "smp handshake for " <> transportName tp
-      liftIO (timeout smpHandshakeTimeout . runExceptT $ smpServerHandshake signKey h ks kh smpServerVRange) >>= \case
+      liftIO (timeout smpHandshakeTimeout . runExceptT $ smpServerHandshake srvCert srvSignKey h ks kh smpServerVRange) >>= \case
         Just (Right th) -> runClientTransport th
         _ -> pure ()
 
@@ -653,9 +647,9 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
               CPSuspend -> withAdminRole $ hPutStrLn h "suspend not implemented"
               CPResume -> withAdminRole $ hPutStrLn h "resume not implemented"
               CPClients -> withAdminRole $ do
-                active <- unliftIO u (asks clients) >>= readTVarIO
+                cls <- getServerClients srv
                 hPutStrLn h "clientId,sessionId,connected,createdAt,rcvActiveAt,sndActiveAt,age,subscriptions"
-                forM_ (IM.toList active) $ \(cid, cl) -> forM_ cl $ \(AClient _ _ Client {sessionId, connected, createdAt, rcvActiveAt, sndActiveAt, subscriptions}) -> do
+                forM_ (IM.toList cls) $ \(cid, (AClient _ _ Client {sessionId, connected, createdAt, rcvActiveAt, sndActiveAt, subscriptions})) -> do
                   connected' <- bshow <$> readTVarIO connected
                   rcvActiveAt' <- strEncode <$> readTVarIO rcvActiveAt
                   sndActiveAt' <- strEncode <$> readTVarIO sndActiveAt
@@ -767,8 +761,8 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
 #else
                   hPutStrLn h "Threads: not available on GHC 8.10"
 #endif
-                  Env {clients, server = Server {subscribers, notifiers, subClients, ntfSubClients}} <- unliftIO u ask
-                  activeClients <- readTVarIO clients
+                  let Server {subscribers, ntfSubscribers} = srv
+                  activeClients <- getServerClients srv
                   hPutStrLn h $ "Clients: " <> show (IM.size activeClients)
                   when (r == CPRAdmin) $ do
                     clQs <- clientTBQueueLengths' activeClients
@@ -782,30 +776,25 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
                     hPutStrLn h $ "Ntf subscriptions (via clients): " <> show ntfSubCnt
                     hPutStrLn h $ "Ntf subscribed clients (via clients): " <> show ntfClCnt
                     hPutStrLn h $ "Ntf subscribed clients queues (via clients, rcvQ, sndQ, msgQ): " <> show ntfClQs
-                  putActiveClientsInfo "SMP" subscribers False
-                  putActiveClientsInfo "Ntf" notifiers True
-                  putSubscribedClients "SMP" subClients False
-                  putSubscribedClients "Ntf" ntfSubClients True
+                  putSubscribersInfo "SMP" subscribers False
+                  putSubscribersInfo "Ntf" ntfSubscribers True
                   where
-                    putActiveClientsInfo :: String -> TMap QueueId (TVar AClient) -> Bool -> IO ()
-                    putActiveClientsInfo protoName clients showIds = do
-                      activeSubs <- readTVarIO clients
+                    putSubscribersInfo :: String -> ServerSubscribers -> Bool -> IO ()
+                    putSubscribersInfo protoName ServerSubscribers {queueSubscribers, subClients} showIds = do
+                      activeSubs <- getSubscribedClients queueSubscribers
                       hPutStrLn h $ protoName <> " subscriptions: " <> show (M.size activeSubs)
                       clnts <- countSubClients activeSubs
                       hPutStrLn h $ protoName <> " subscribed clients: " <> show (IS.size clnts) <> (if showIds then " " <> show (IS.toList clnts) else "")
+                      clnts' <- readTVarIO subClients
+                      hPutStrLn h $ protoName <> " subscribed clients count 2: " <> show (IS.size clnts') <> (if showIds then " " <> show clnts' else "")
                       where
-                        countSubClients :: M.Map QueueId (TVar AClient) -> IO IS.IntSet
-                        countSubClients = foldM (\ !s c -> (`IS.insert` s) . clientId' <$> readTVarIO c) IS.empty
-                    putSubscribedClients :: String -> TVar (IM.IntMap AClient) -> Bool -> IO ()
-                    putSubscribedClients protoName subClnts showIds = do
-                      clnts <- readTVarIO subClnts
-                      hPutStrLn h $ protoName <> " subscribed clients count 2: " <> show (IM.size clnts) <> (if showIds then " " <> show (IM.keys clnts) else "")
-                    countClientSubs :: (forall s. Client s -> TMap QueueId a) -> Maybe (M.Map QueueId a -> IO (Int, Int, Int, Int)) -> IM.IntMap (Maybe AClient) -> IO (Int, (Int, Int, Int, Int), Int, (Natural, Natural, Natural))
+                        countSubClients :: M.Map QueueId (TVar (Maybe AClient)) -> IO IS.IntSet
+                        countSubClients = foldM (\ !s c -> maybe s ((`IS.insert` s) . clientId') <$> readTVarIO c) IS.empty
+                    countClientSubs :: (forall s. Client s -> TMap QueueId a) -> Maybe (M.Map QueueId a -> IO (Int, Int, Int, Int)) -> IM.IntMap AClient -> IO (Int, (Int, Int, Int, Int), Int, (Natural, Natural, Natural))
                     countClientSubs subSel countSubs_ = foldM addSubs (0, (0, 0, 0, 0), 0, (0, 0, 0))
                       where
-                        addSubs :: (Int, (Int, Int, Int, Int), Int, (Natural, Natural, Natural)) -> Maybe AClient -> IO (Int, (Int, Int, Int, Int), Int, (Natural, Natural, Natural))
-                        addSubs acc Nothing = pure acc
-                        addSubs (!subCnt, cnts@(!c1, !c2, !c3, !c4), !clCnt, !qs) (Just acl@(AClient _ _ cl)) = do
+                        addSubs :: (Int, (Int, Int, Int, Int), Int, (Natural, Natural, Natural)) -> AClient -> IO (Int, (Int, Int, Int, Int), Int, (Natural, Natural, Natural))
+                        addSubs (!subCnt, cnts@(!c1, !c2, !c3, !c4), !clCnt, !qs) acl@(AClient _ _ cl) = do
                           subs <- readTVarIO $ subSel cl
                           cnts' <- case countSubs_ of
                             Nothing -> pure cnts
@@ -816,8 +805,8 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
                               clCnt' = if cnt == 0 then clCnt else clCnt + 1
                           qs' <- if cnt == 0 then pure qs else addQueueLengths qs acl
                           pure (subCnt + cnt, cnts', clCnt', qs')
-                    clientTBQueueLengths' :: Foldable t => t (Maybe AClient) -> IO (Natural, Natural, Natural)
-                    clientTBQueueLengths' = foldM (\acc -> maybe (pure acc) (addQueueLengths acc)) (0, 0, 0)
+                    clientTBQueueLengths' :: Foldable t => t AClient -> IO (Natural, Natural, Natural)
+                    clientTBQueueLengths' = foldM addQueueLengths (0, 0, 0)
                     addQueueLengths (!rl, !sl, !ml) (AClient _ _ cl) = do
                       (rl', sl', ml') <- queueLengths cl
                       pure (rl + rl', sl + sl', ml + ml')
@@ -896,30 +885,28 @@ runClientTransport :: Transport c => THandleSMP c 'TServer -> M ()
 runClientTransport h@THandle {params = thParams@THandleParams {thVersion, sessionId}} = do
   q <- asks $ tbqSize . config
   ts <- liftIO getSystemTime
-  active <- asks clients
   nextClientId <- asks clientSeq
   clientId <- atomically $ stateTVar nextClientId $ \next -> (next, next + 1)
-  atomically $ modifyTVar' active $ IM.insert clientId Nothing
   AMS qt mt ms <- asks msgStore
   c <- liftIO $ newClient qt mt clientId q thVersion sessionId ts
-  runClientThreads qt mt ms active c clientId `finally` clientDisconnected c
+  runClientThreads qt mt ms c `finally` clientDisconnected c
   where
-    runClientThreads :: MsgStoreClass (MsgStore qs ms) => SQSType qs -> SMSType ms -> MsgStore qs ms -> TVar (IM.IntMap (Maybe AClient)) -> Client (MsgStore qs ms) -> IS.Key -> M ()
-    runClientThreads qt mt ms active c clientId = do
-      atomically $ modifyTVar' active $ IM.insert clientId $ Just (AClient qt mt c)
+    runClientThreads :: MsgStoreClass (MsgStore qs ms) => SQSType qs -> SMSType ms -> MsgStore qs ms -> Client (MsgStore qs ms) -> M ()
+    runClientThreads qt mt ms c = do
       s <- asks server
-      expCfg <- asks $ inactiveClientExpiration . config
-      th <- newMVar h -- put TH under a fair lock to interleave messages and command responses
-      labelMyThread . B.unpack $ "client $" <> encode sessionId
-      raceAny_ $ [liftIO $ send th c, liftIO $ sendMsg th c, client thParams s ms c, receive h ms c] <> disconnectThread_ c s expCfg
+      whenM (liftIO $ insertServerClient (AClient qt mt c) s) $ do
+        expCfg <- asks $ inactiveClientExpiration . config
+        th <- newMVar h -- put TH under a fair lock to interleave messages and command responses
+        labelMyThread . B.unpack $ "client $" <> encode sessionId
+        raceAny_ $ [liftIO $ send th c, liftIO $ sendMsg th c, client thParams s ms c, receive h ms c] <> disconnectThread_ c s expCfg
     disconnectThread_ :: Client s -> Server -> Maybe ExpirationConfig -> [M ()]
     disconnectThread_ c s (Just expCfg) = [liftIO $ disconnectTransport h (rcvActiveAt c) (sndActiveAt c) expCfg (noSubscriptions c s)]
     disconnectThread_ _ _ _ = []
-    noSubscriptions Client {clientId} s = do
-      hasSubs <- IM.member clientId <$> readTVarIO (subClients s)
+    noSubscriptions Client {clientId} Server {subscribers, ntfSubscribers} = do
+      hasSubs <- IS.member clientId <$> readTVarIO (subClients subscribers)
       if hasSubs
         then pure False
-        else not . IM.member clientId <$> readTVarIO (ntfSubClients s)
+        else not . IS.member clientId <$> readTVarIO (subClients ntfSubscribers)
 
 clientDisconnected :: Client s -> M ()
 clientDisconnected c@Client {clientId, subscriptions, ntfSubscriptions, connected, sessionId, endThreads} = do
@@ -931,26 +918,17 @@ clientDisconnected c@Client {clientId, subscriptions, ntfSubscriptions, connecte
   ntfSubs <- atomically $ swapTVar ntfSubscriptions M.empty
   liftIO $ mapM_ cancelSub subs
   whenM (asks serverActive >>= readTVarIO) $ do
-    Server {subscribers, notifiers, subClients, ntfSubClients} <- asks server
+    srv@Server {subscribers, ntfSubscribers} <- asks server
     liftIO $ updateSubscribers subs subscribers
-    liftIO $ updateSubscribers ntfSubs notifiers
-    asks clients >>= atomically . (`modifyTVar'` IM.delete clientId)
-    atomically $ modifyTVar' subClients $ IM.delete clientId
-    atomically $ modifyTVar' ntfSubClients $ IM.delete clientId
+    liftIO $ updateSubscribers ntfSubs ntfSubscribers
+    liftIO $ deleteServerClient clientId srv
   tIds <- atomically $ swapTVar endThreads IM.empty
   liftIO $ mapM_ (mapM_ killThread <=< deRefWeak) tIds
   where
-    updateSubscribers :: M.Map QueueId a -> TMap QueueId (TVar AClient) -> IO ()
-    updateSubscribers subs srvSubs =
-      forM_ (M.keys subs) $ \qId ->
-        -- lookup of the subscribed client TVar can be in separate transaction,
-        -- as long as the client is read in the same transaction -
-        -- it prevents removing the next subscribed client.
-        TM.lookupIO qId srvSubs >>=
-          mapM_ (\c' -> atomically $ whenM (sameClientId c <$> readTVar c') $ TM.delete qId srvSubs)
-
-sameClientId :: Client s -> AClient -> Bool
-sameClientId Client {clientId} ac = clientId == clientId' ac
+    updateSubscribers :: M.Map QueueId a -> ServerSubscribers -> IO ()
+    updateSubscribers subs ServerSubscribers {queueSubscribers, subClients} = do
+      mapM_ (\qId -> deleteSubcribedClient qId c queueSubscribers) (M.keys subs)
+      atomically $ modifyTVar' subClients $ IS.delete clientId
 
 cancelSub :: Sub -> IO ()
 cancelSub s = case subThread s of
@@ -1071,7 +1049,7 @@ verifyTransmission ms auth_ tAuth authorized queueId cmd =
     Cmd SSender PING -> pure $ VRVerified Nothing
     Cmd SSender RFWD {} -> pure $ VRVerified Nothing
     Cmd SSenderLink (LKEY k) -> verifySecure SSenderLink k
-    Cmd SSenderLink LGET -> verifyQueue (\q -> if isContact (snd q) then VRVerified (Just q) else VRFailed) <$> get SSenderLink
+    Cmd SSenderLink LGET -> verifyQueue (\q -> if isContactQueue (snd q) then VRVerified (Just q) else VRFailed) <$> get SSenderLink
     -- NSUB will not be accepted without authorization
     Cmd SNotifier NSUB -> verifyQueue (\q -> maybe dummyVerify (\n -> Just q `verifiedWith` notifierKey n) (notifier $ snd q)) <$> get SNotifier
     Cmd SProxiedClient _ -> pure $ VRVerified Nothing
@@ -1089,12 +1067,15 @@ verifyTransmission ms auth_ tAuth authorized queueId cmd =
     allowedKey k = \case
       QueueRec {queueMode = Just QMMessaging, senderKey} -> maybe True (k ==) senderKey
       _ -> False
-    isContact = \case
-      QueueRec {queueMode = Just QMContact} -> True
-      _ -> False
     get :: DirectParty p => SParty p -> M (Either ErrorType (StoreQueue s, QueueRec))
     get party = liftIO $ getQueueRec ms party queueId
 
+isContactQueue :: QueueRec -> Bool
+isContactQueue QueueRec {queueMode, senderKey} = case queueMode of
+  Just QMMessaging -> False
+  Just QMContact -> True
+  Nothing -> isNothing senderKey -- for backward compatibility with pre-SKEY contact addresses
+
 verifyCmdAuthorization :: Maybe (THandleAuth 'TServer, C.CbNonce) -> Maybe TransmissionAuth -> ByteString -> C.APublicAuthKey -> Bool
 verifyCmdAuthorization auth_ tAuth authorized key = maybe False (verify key) tAuth
   where
@@ -1151,7 +1132,7 @@ forkClient Client {endThreads, endThreadSeq} label action = do
 client :: forall s. MsgStoreClass s => THandleParams SMPVersion 'TServer -> Server -> s -> Client s -> M ()
 client
   thParams'
-  Server {subscribedQ, ntfSubscribedQ, subscribers}
+  Server {subscribers, ntfSubscribers}
   ms
   clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId, procThreads} = do
     labelMyThread . B.unpack $ "client $" <> encode sessionId <> " commands"
@@ -1253,7 +1234,7 @@ client
         RFWD encBlock -> (corrId, NoEntity,) <$> processForwardedCommand encBlock
       Cmd SSenderLink command -> Just <$> case command of
         LKEY k -> withQueue $ \q qr -> checkMode QMMessaging qr $ secureQueue_ q k $>> getQueueLink_ q qr
-        LGET -> withQueue $ \q qr -> checkMode QMContact qr $ getQueueLink_ q qr
+        LGET -> withQueue $ \q qr -> checkContact qr $ getQueueLink_ q qr
       Cmd SNotifier NSUB -> Just <$> subscribeNotifications
       Cmd SRecipient command ->
         Just <$> case command of
@@ -1269,11 +1250,11 @@ client
           KEY sKey -> withQueue $ \q _ -> either err (corrId,entId,) <$> secureQueue_ q sKey
           RKEY rKeys -> withQueue $ \q qr -> checkMode QMContact qr $ OK <$$ liftIO (updateKeys (queueStore ms) q rKeys)
           LSET lnkId d ->
-            withQueue $ \q qr -> checkMode QMContact qr $ liftIO $ case queueData qr of
+            withQueue $ \q qr -> checkContact qr $ liftIO $ case queueData qr of
               Just (lnkId', _) | lnkId' /= lnkId -> pure $ Left AUTH
               _ -> OK <$$ addQueueLinkData (queueStore ms) q lnkId d
           LDEL ->
-            withQueue $ \q qr -> checkMode QMContact qr $ liftIO $ case queueData qr of
+            withQueue $ \q qr -> checkContact qr $ liftIO $ case queueData qr of
               Just _ -> OK <$$ deleteQueueLinkData (queueStore ms) q
               Nothing -> pure $ Right OK
           NKEY nKey dhKey -> withQueue $ \q _ -> addQueueNotifier_ q nKey dhKey
@@ -1343,6 +1324,13 @@ client
                         pure $ IDS QIK {rcvId, sndId, rcvPublicDhKey, queueMode, linkId = fst <$> queueData} -- , serverNtfCreds = snd <$> ntf
           (corrId,entId,) <$> tryCreate (3 :: Int)
 
+        -- this check allows to support contact queues created prior to SKEY,
+        -- using `queueMode == Just QMContact` would prevent it, as they have queueMode `Nothing`.
+        checkContact :: QueueRec -> M (Either ErrorType BrokerMsg) -> M (Transmission BrokerMsg)
+        checkContact qr a =
+          either err (corrId,entId,)
+            <$> if isContactQueue qr then a else pure $ Left AUTH
+
         checkMode :: QueueMode -> QueueRec -> M (Either ErrorType BrokerMsg) -> M (Transmission BrokerMsg)
         checkMode qm QueueRec {queueMode} a =
           either err (corrId,entId,)
@@ -1372,7 +1360,7 @@ client
                 Left e -> pure $ ERR e
                 Right nId_ -> do
                   incStat . ntfCreated =<< asks serverStats
-                  forM_ nId_ $ \nId -> atomically $ writeTQueue ntfSubscribedQ (nId, clientId, False)
+                  forM_ nId_ $ \nId -> atomically $ writeTQueue (subQ ntfSubscribers) (nId, clientId, False)
                   pure $ NID notifierId rcvPublicDhKey
 
         deleteQueueNotifier_ :: StoreQueue s -> M (Transmission BrokerMsg)
@@ -1383,7 +1371,7 @@ client
               stats <- asks serverStats
               deleted <- asks ntfStore >>= liftIO . (`deleteNtfs` nId)
               when (deleted > 0) $ liftIO $ atomicModifyIORef'_ (ntfCount stats) (subtract deleted)
-              atomically $ writeTQueue ntfSubscribedQ (nId, clientId, False)
+              atomically $ writeTQueue (subQ ntfSubscribers) (nId, clientId, False)
               incStat $ ntfDeleted stats
               pure ok
             Right Nothing -> pure ok
@@ -1394,7 +1382,7 @@ client
 
         subscribeQueue :: StoreQueue s -> QueueRec -> M (Transmission BrokerMsg)
         subscribeQueue q qr =
-          atomically (TM.lookup rId subscriptions) >>= \case
+          liftIO (TM.lookupIO rId subscriptions) >>= \case
             Nothing -> newSub >>= deliver True
             Just s@Sub {subThread} -> do
               stats <- asks serverStats
@@ -1410,7 +1398,7 @@ client
             rId = recipientId q
             newSub :: M Sub
             newSub = time "SUB newSub" . atomically $ do
-              writeTQueue subscribedQ (rId, clientId, True)
+              writeTQueue (subQ subscribers) (rId, clientId, True)
               sub <- newSubscription NoSub
               TM.insert rId sub subscriptions
               pure sub
@@ -1486,7 +1474,7 @@ client
           pure ok
           where
             newSub = do
-              writeTQueue ntfSubscribedQ (entId, clientId, True)
+              writeTQueue (subQ ntfSubscribers) (entId, clientId, True)
               TM.insert entId () ntfSubscriptions
 
         acknowledgeMsg :: MsgId -> StoreQueue s -> QueueRec -> M (Transmission BrokerMsg)
@@ -1522,7 +1510,7 @@ client
                 incStat $ msgRecv stats
                 if isGet
                   then incStat $ msgRecvGet stats
-                  else pure () -- TODO skip notification delivery for delivered message  
+                  else pure () -- TODO skip notification delivery for delivered message
                   -- skipping delivery fails tests, it should be counted in msgNtfSkipped
                   -- forM_ (notifierId <$> notifier qr) $ \nId -> do
                   --   ns <- asks ntfStore
@@ -1595,18 +1583,19 @@ client
             -- - nothing was delivered to this subscription (to avoid race conditions with the recipient).
             tryDeliverMessage :: Message -> IO ()
             tryDeliverMessage msg =
-              -- the subscription is checked outside of STM to avoid transaction cost
+              -- the subscribed client var is read outside of STM to avoid transaction cost
               -- in case no client is subscribed.
-              whenM (TM.memberIO rId subscribers) $
-                atomically deliverToSub >>= mapM_ forkDeliver
+              getSubscribedClient rId (queueSubscribers subscribers)
+                $>>= atomically . deliverToSub
+                >>= mapM_ forkDeliver
               where
                 rId = recipientId q
-                deliverToSub =
-                  -- lookup has ot be in the same transaction,
+                deliverToSub rcv =
+                  -- reading client TVar in the same transaction,
                   -- so that if subscription ends, it re-evalutates
                   -- and delivery is cancelled -
                   -- the new client will receive message in response to SUB.
-                  (TM.lookup rId subscribers >>= mapM readTVar)
+                  readTVar rcv
                     $>>= \rc@(AClient _ _ Client {subscriptions = subs, sndQ = sndQ'}) -> TM.lookup rId subs
                     $>>= \s@Sub {subThread, delivered} -> case subThread of
                       ProhibitSub -> pure Nothing
@@ -1635,9 +1624,9 @@ client
                       labelMyThread $ B.unpack ("client $" <> encode sessionId) <> " deliver/SEND"
                       -- lookup can be outside of STM transaction,
                       -- as long as the check that it is the same client is inside.
-                      TM.lookupIO rId subscribers >>= mapM_ deliverIfSame
-                    deliverIfSame rc' = time "deliver" . atomically $
-                      whenM (sameClientId rc <$> readTVar rc') $
+                      getSubscribedClient rId (queueSubscribers subscribers) >>= mapM_ deliverIfSame
+                    deliverIfSame rcv = time "deliver" . atomically $
+                      whenM (sameClient rc rcv) $
                         tryTakeTMVar delivered >>= \case
                           Just _ -> pure () -- if a message was already delivered, should not deliver more
                           Nothing -> do
@@ -1750,7 +1739,7 @@ client
             Right qr -> do
               -- Possibly, the same should be done if the queue is suspended, but currently we do not use it
               atomically $ do
-                writeTQueue subscribedQ (entId, clientId, False)
+                writeTQueue (subQ subscribers) (entId, clientId, False)
                 -- queue is usually deleted by the same client that is currently subscribed,
                 -- we delete subscription here, so the client with no subscriptions can be disconnected.
                 TM.delete entId subscriptions
@@ -1760,7 +1749,7 @@ client
                 stats <- asks serverStats
                 deleted <- asks ntfStore >>= liftIO . (`deleteNtfs` nId)
                 when (deleted > 0) $ liftIO $ atomicModifyIORef'_ (ntfCount stats) (subtract deleted)
-                atomically $ writeTQueue ntfSubscribedQ (nId, clientId, False)
+                atomically $ writeTQueue (subQ ntfSubscribers) (nId, clientId, False)
               updateDeletedStats qr
               pure ok
             Left e -> pure $ err e
@@ -1985,7 +1974,7 @@ restoreServerNtfs =
             renameFile f $ f <> ".bak"
             let NtfStore ns' = ns
             storedQueues <- M.size <$> readTVarIO ns'
-            logNote $ "notifications restored, " <> tshow lineCount <> " lines processed" 
+            logNote $ "notifications restored, " <> tshow lineCount <> " lines processed"
             pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues}
       where
         restoreNtf :: NtfStore -> Int64 -> (Int, Int, Int) -> LB.ByteString -> ExceptT String IO (Int, Int, Int)

src/Simplex/Messaging/Server/CLI.hs

@@ -34,7 +34,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (ProtoServerWithAuth (..), ProtocolServer (..), ProtocolTypeI)
 import Simplex.Messaging.Server.Env.STM (AServerStoreCfg (..), ServerStoreCfg (..), StartOptions (..), StorePaths (..))
 import Simplex.Messaging.Server.QueueStore.Postgres.Config (PostgresStoreCfg (..))
-import Simplex.Messaging.Transport (ATransport (..), TLS, Transport (..))
+import Simplex.Messaging.Transport (ASrvTransport, ATransport (..), TLS, Transport (..))
 import Simplex.Messaging.Transport.Server (AddHTTP, loadFileFingerprint)
 import Simplex.Messaging.Transport.WebSockets (WS)
 import Simplex.Messaging.Util (eitherToMaybe, whenM)
@@ -363,7 +363,7 @@ checkSavedFingerprint cfgPath x509cfg = do
   where
     c = combine cfgPath . ($ x509cfg)
 
-iniTransports :: Ini -> [(ServiceName, ATransport, AddHTTP)]
+iniTransports :: Ini -> [(ServiceName, ASrvTransport, AddHTTP)]
 iniTransports ini =
   let smpPorts = ports $ strictIni "TRANSPORT" "port" ini
       ws = strictIni "TRANSPORT" "websockets" ini
@@ -373,7 +373,7 @@ iniTransports ini =
         | otherwise = ports ws \\ smpPorts
    in ts (transport @TLS) smpPorts <> ts (transport @WS) wsPorts
   where
-    ts :: ATransport -> [ServiceName] -> [(ServiceName, ATransport, AddHTTP)]
+    ts :: ASrvTransport -> [ServiceName] -> [(ServiceName, ASrvTransport, AddHTTP)]
     ts t = map (\port -> (port, t, webPort == Just port))
     webPort = T.unpack <$> eitherToMaybe (lookupValue "WEB" "https" ini)
     ports = map T.unpack . T.splitOn ","
@@ -387,14 +387,14 @@ iniDBOptions ini _default@DBOpts {connstr, schema, poolSize} =
       createSchema = False
     }
 
-printServerConfig :: String -> [(ServiceName, ATransport, AddHTTP)] -> Maybe FilePath -> IO ()
+printServerConfig :: String -> [(ServiceName, ASrvTransport, AddHTTP)] -> Maybe FilePath -> IO ()
 printServerConfig protocol transports logFile = do
   putStrLn $ case logFile of
     Just f -> "Store log: " <> f
     _ -> "Store log disabled."
   printServerTransports protocol transports
 
-printServerTransports :: String -> [(ServiceName, ATransport, AddHTTP)] -> IO ()
+printServerTransports :: String -> [(ServiceName, ASrvTransport, AddHTTP)] -> IO ()
 printServerTransports protocol ts = do
   forM_ ts $ \(p, ATransport t, addHTTP) -> do
     let descr = p <> " (" <> transportName t <> ")..."
@@ -405,7 +405,7 @@ printServerTransports protocol ts = do
       "\nWARNING: the clients will use port 443 by default soon.\n\
       \Set `port` in smp-server.ini section [TRANSPORT] to `5223,443`\n"
 
-printSMPServerConfig :: [(ServiceName, ATransport, AddHTTP)] -> AServerStoreCfg -> IO ()
+printSMPServerConfig :: [(ServiceName, ASrvTransport, AddHTTP)] -> AServerStoreCfg -> IO ()
 printSMPServerConfig transports (ASSCfg _ _ cfg) = case cfg of
   SSCMemory sp_ -> printServerConfig "SMP" transports $ (\StorePaths {storeLogFile} -> storeLogFile) <$> sp_
   SSCMemoryJournal {storeLogFile} -> printServerConfig "SMP" transports $ Just storeLogFile

src/Simplex/Messaging/Server/Env/STM.hs

@@ -18,7 +18,59 @@
 #endif
 {-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
 
-module Simplex.Messaging.Server.Env.STM where
+module Simplex.Messaging.Server.Env.STM
+  ( ServerConfig (..),
+    ServerStoreCfg (..),
+    AServerStoreCfg (..),
+    StorePaths (..),
+    StartOptions (..),
+    Env (..),
+    Server (..),
+    ServerSubscribers (..),
+    SubscribedClients,
+    ProxyAgent (..),
+    Client (..),
+    AClient (..),
+    ClientId,
+    Subscribed,
+    Sub (..),
+    ServerSub (..),
+    SubscriptionThread (..),
+    MsgStore,
+    AMsgStore (..),
+    AStoreType (..),
+    newEnv,
+    mkJournalStoreConfig,
+    newClient,
+    getServerClients,
+    getServerClient,
+    insertServerClient,
+    deleteServerClient,
+    getSubscribedClients,
+    getSubscribedClient,
+    upsertSubscribedClient,
+    lookupDeleteSubscribedClient,
+    deleteSubcribedClient,
+    sameClientId,
+    sameClient,
+    clientId',
+    newSubscription,
+    newProhibitedSub,
+    defaultMsgQueueQuota,
+    defMsgExpirationDays,
+    defNtfExpirationHours,
+    defaultMessageExpiration,
+    defaultNtfExpiration,
+    defaultInactiveClientExpiration,
+    defaultProxyClientConcurrency,
+    defaultMaxJournalMsgCount,
+    defaultMaxJournalStateLines,
+    defaultIdleQueueInterval,
+    journalMsgStoreDepth,
+    readWriteQueueStore,
+    noPostgresExit,
+  )
+where
 
 import Control.Concurrent (ThreadId)
 import Control.Logger.Simple
@@ -29,9 +81,12 @@ import Data.ByteString.Char8 (ByteString)
 import Data.Int (Int64)
 import Data.IntMap.Strict (IntMap)
 import qualified Data.IntMap.Strict as IM
+import Data.IntSet (IntSet)
+import qualified Data.IntSet as IS
 import Data.Kind (Constraint)
 import Data.List (intercalate)
 import Data.List.NonEmpty (NonEmpty)
+import Data.Map.Strict (Map)
 import Data.Maybe (isJust)
 import qualified Data.Text as T
 import Data.Time.Clock (getCurrentTime, nominalDay)
@@ -64,8 +119,9 @@ import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.Server.StoreLog.ReadWrite
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Transport (ATransport, VersionRangeSMP, VersionSMP)
+import Simplex.Messaging.Transport (ASrvTransport, VersionRangeSMP, VersionSMP)
 import Simplex.Messaging.Transport.Server
+import Simplex.Messaging.Util (ifM, whenM, ($>>=))
 import System.Directory (doesFileExist)
 import System.Exit (exitFailure)
 import System.IO (IOMode (..))
@@ -73,7 +129,7 @@ import System.Mem.Weak (Weak)
 import UnliftIO.STM
 
 data ServerConfig = ServerConfig
-  { transports :: [(ServiceName, ATransport, AddHTTP)],
+  { transports :: [(ServiceName, ASrvTransport, AddHTTP)],
     smpHandshakeTimeout :: Int,
     tbqSize :: Natural,
     msgQueueQuota :: Int,
@@ -203,7 +259,6 @@ data Env = Env
     serverStats :: ServerStats,
     sockets :: TVar [(ServiceName, SocketState)],
     clientSeq :: TVar ClientId,
-    clients :: TVar (IntMap (Maybe AClient)),
     proxyAgent :: ProxyAgent -- senders served on this proxy
   }
 
@@ -236,17 +291,72 @@ data AMsgStore =
 type Subscribed = Bool
 
 data Server = Server
-  { subscribedQ :: TQueue (RecipientId, ClientId, Subscribed),
-    subscribers :: TMap RecipientId (TVar AClient),
-    ntfSubscribedQ :: TQueue (NotifierId, ClientId, Subscribed),
-    notifiers :: TMap NotifierId (TVar AClient),
-    subClients :: TVar (IntMap AClient), -- clients with SMP subscriptions
-    ntfSubClients :: TVar (IntMap AClient), -- clients with Ntf subscriptions
-    pendingSubEvents :: TVar (IntMap (NonEmpty (RecipientId, Subscribed))),
-    pendingNtfSubEvents :: TVar (IntMap (NonEmpty (NotifierId, Subscribed))),
+  { clients :: ServerClients,
+    subscribers :: ServerSubscribers,
+    ntfSubscribers :: ServerSubscribers,
     savingLock :: Lock
   }
 
+-- not exported, to prevent concurrent IntMap lookups inside STM transactions.
+newtype ServerClients = ServerClients {serverClients :: TVar (IntMap AClient)}
+
+data ServerSubscribers = ServerSubscribers
+  { subQ :: TQueue (QueueId, ClientId, Subscribed),
+    queueSubscribers :: SubscribedClients,
+    subClients :: TVar IntSet,
+    pendingEvents :: TVar (IntMap (NonEmpty (EntityId, BrokerMsg)))
+  }
+
+-- not exported, to prevent accidental concurrent Map lookups inside STM transactions.
+-- Map stores TVars with pointers to the clients rather than client ID to allow reading the same TVar
+-- inside transactions to ensure that transaction is re-evaluated in case subscriber changes.
+-- Storing Maybe allows to have continuity of subscription when the same user client disconnects and re-connects -
+-- any STM transaction that reads subscribed client will re-evaluate in this case.
+-- The subscriptions that were made at any point are not removed -
+-- this is a better trade-off with intermittently connected mobile clients.
+data SubscribedClients = SubscribedClients (TMap EntityId (TVar (Maybe AClient)))
+
+getSubscribedClients :: SubscribedClients -> IO (Map EntityId (TVar (Maybe AClient)))
+getSubscribedClients (SubscribedClients cs) = readTVarIO cs
+
+getSubscribedClient :: EntityId -> SubscribedClients -> IO (Maybe (TVar (Maybe AClient)))
+getSubscribedClient entId (SubscribedClients cs) = TM.lookupIO entId cs
+{-# INLINE getSubscribedClient #-}
+
+-- insert subscribed and current client, return previously subscribed client if it is different
+upsertSubscribedClient :: EntityId -> AClient -> SubscribedClients -> STM (Maybe AClient)
+upsertSubscribedClient entId ac@(AClient _ _ c) (SubscribedClients cs) =
+  TM.lookup entId cs >>= \case
+    Nothing -> Nothing <$ TM.insertM entId (newTVar (Just ac)) cs
+    Just cv ->
+      readTVar cv >>= \case
+        Just c' | sameClientId c c' -> pure Nothing
+        c_ -> c_ <$ writeTVar cv (Just ac)
+
+-- lookup and delete currently subscribed client
+lookupDeleteSubscribedClient :: EntityId -> SubscribedClients -> STM (Maybe AClient)
+lookupDeleteSubscribedClient entId (SubscribedClients cs) =
+  TM.lookupDelete entId cs $>>= (`swapTVar` Nothing)
+
+deleteSubcribedClient :: EntityId -> Client s -> SubscribedClients -> IO ()
+deleteSubcribedClient entId c (SubscribedClients cs) =
+  -- lookup of the subscribed client TVar can be in separate transaction,
+  -- as long as the client is read in the same transaction -
+  -- it prevents removing the next subscribed client and also avoids STM contention for the Map.
+  TM.lookupIO entId cs >>= mapM_ (\cv -> atomically $ whenM (sameClient c cv) $ delete cv)
+  where
+    delete cv = do
+      writeTVar cv Nothing
+      TM.delete entId cs
+
+sameClientId :: Client s -> AClient -> Bool
+sameClientId Client {clientId} ac = clientId == clientId' ac
+{-# INLINE sameClientId #-}
+
+sameClient :: Client s -> TVar (Maybe AClient) -> STM Bool
+sameClient c cv = maybe False (sameClientId c) <$> readTVar cv
+{-# INLINE sameClient #-}
+
 newtype ProxyAgent = ProxyAgent
   { smpAgent :: SMPClientAgent
   }
@@ -288,16 +398,40 @@ data Sub = Sub
 
 newServer :: IO Server
 newServer = do
-  subscribedQ <- newTQueueIO
-  subscribers <- TM.emptyIO
-  ntfSubscribedQ <- newTQueueIO
-  notifiers <- TM.emptyIO
-  subClients <- newTVarIO IM.empty
-  ntfSubClients <- newTVarIO IM.empty
-  pendingSubEvents <- newTVarIO IM.empty
-  pendingNtfSubEvents <- newTVarIO IM.empty
+  clients <- ServerClients <$> newTVarIO mempty
+  subscribers <- newServerSubscribers
+  ntfSubscribers <- newServerSubscribers
   savingLock <- createLockIO
-  return Server {subscribedQ, subscribers, ntfSubscribedQ, notifiers, subClients, ntfSubClients, pendingSubEvents, pendingNtfSubEvents, savingLock}
+  return Server {clients, subscribers, ntfSubscribers, savingLock}
+
+getServerClients :: Server -> IO (IntMap AClient)
+getServerClients = readTVarIO . serverClients . clients
+{-# INLINE getServerClients #-}
+
+getServerClient :: ClientId -> Server -> IO (Maybe AClient)
+getServerClient cId s = IM.lookup cId <$> getServerClients s
+{-# INLINE getServerClient #-}
+
+insertServerClient :: AClient -> Server -> IO Bool
+insertServerClient ac@(AClient _ _ Client {clientId, connected}) Server {clients} =
+  atomically $
+    ifM
+      (readTVar connected)
+      (True <$ modifyTVar' (serverClients clients) (IM.insert clientId ac))
+      (pure False)
+{-# INLINE insertServerClient #-}
+
+deleteServerClient :: ClientId -> Server -> IO ()
+deleteServerClient cId Server {clients} = atomically $ modifyTVar' (serverClients clients) $ IM.delete cId
+{-# INLINE deleteServerClient #-}
+
+newServerSubscribers :: IO ServerSubscribers
+newServerSubscribers = do
+  subQ <- newTQueueIO
+  queueSubscribers <- SubscribedClients <$> TM.emptyIO
+  subClients <- newTVarIO IS.empty
+  pendingEvents <- newTVarIO IM.empty
+  pure ServerSubscribers {subQ, queueSubscribers, subClients, pendingEvents}
 
 newClient :: SQSType qs -> SMSType ms -> ClientId -> Natural -> VersionSMP -> ByteString -> SystemTime -> IO (Client (MsgStore qs ms))
 newClient _ _ clientId qSize thVersion sessionId createdAt = do
@@ -312,7 +446,24 @@ newClient _ _ clientId qSize thVersion sessionId createdAt = do
   connected <- newTVarIO True
   rcvActiveAt <- newTVarIO createdAt
   sndActiveAt <- newTVarIO createdAt
-  return Client {clientId, subscriptions, ntfSubscriptions, rcvQ, sndQ, msgQ, procThreads, endThreads, endThreadSeq, thVersion, sessionId, connected, createdAt, rcvActiveAt, sndActiveAt}
+  return
+    Client
+      { clientId,
+        subscriptions,
+        ntfSubscriptions,
+        rcvQ,
+        sndQ,
+        msgQ,
+        procThreads,
+        endThreads,
+        endThreadSeq,
+        thVersion,
+        sessionId,
+        connected,
+        createdAt,
+        rcvActiveAt,
+        sndActiveAt
+      }
 
 newSubscription :: SubscriptionThread -> STM Sub
 newSubscription st = do
@@ -362,9 +513,24 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smp
   serverStats <- newServerStats =<< getCurrentTime
   sockets <- newTVarIO []
   clientSeq <- newTVarIO 0
-  clients <- newTVarIO mempty
   proxyAgent <- newSMPProxyAgent smpAgentCfg random
-  pure Env {serverActive, config, serverInfo, server, serverIdentity, msgStore, ntfStore, random, tlsServerCreds, httpServerCreds, serverStats, sockets, clientSeq, clients, proxyAgent}
+  pure
+    Env
+      { serverActive,
+        config,
+        serverInfo,
+        server,
+        serverIdentity,
+        msgStore,
+        ntfStore,
+        random,
+        tlsServerCreds,
+        httpServerCreds,
+        serverStats,
+        sockets,
+        clientSeq,
+        proxyAgent
+      }
   where
     loadStoreLog :: StoreQueueClass q => (RecipientId -> QueueRec -> IO q) -> FilePath -> STMQueueStore q -> IO ()
     loadStoreLog mkQ f st = do

src/Simplex/Messaging/Server/Prometheus.hs

@@ -33,13 +33,16 @@ data RealTimeMetrics = RealTimeMetrics
   { socketStats :: [(ServiceName, SocketStats)],
     threadsCount :: Int,
     clientsCount :: Int,
-    smpSubsCount :: Int,
-    smpSubClientsCount :: Int,
-    ntfSubsCount :: Int,
-    ntfSubClientsCount :: Int,
+    smpSubs :: RTSubscriberMetrics,
+    ntfSubs :: RTSubscriberMetrics,
     loadedCounts :: LoadedQueueCounts
   }
 
+data RTSubscriberMetrics = RTSubscriberMetrics
+  { subsCount :: Int,
+    subClientsCount :: Int
+  }
+
 {-# FOURMOLU_DISABLE\n#-}
 prometheusMetrics :: ServerMetrics -> RealTimeMetrics -> UTCTime -> Text
 prometheusMetrics sm rtm ts =
@@ -50,10 +53,8 @@ prometheusMetrics sm rtm ts =
       { socketStats,
         threadsCount,
         clientsCount,
-        smpSubsCount,
-        smpSubClientsCount,
-        ntfSubsCount,
-        ntfSubClientsCount,
+        smpSubs,
+        ntfSubs,
         loadedCounts
       } = rtm
     ServerStatsData
@@ -367,21 +368,21 @@ prometheusMetrics sm rtm ts =
       \# TYPE simplex_smp_clients_total gauge\n\
       \simplex_smp_clients_total " <> mshow clientsCount <> "\n\
       \\n\
-      \# HELP simplex_smp_subscribtion_total Total subscriptions\n\
+      \# HELP simplex_smp_subscribtion_total Total SMP subscriptions\n\
       \# TYPE simplex_smp_subscribtion_total gauge\n\
-      \simplex_smp_subscribtion_total " <> mshow smpSubsCount <> "\n# smpSubs\n\
+      \simplex_smp_subscribtion_total " <> mshow (subsCount smpSubs) <> "\n# smp.subsCount\n\
       \\n\
-      \# HELP simplex_smp_subscribtion_clients_total Subscribed clients, first counting method\n\
+      \# HELP simplex_smp_subscribtion_clients_total Subscribed clients\n\
       \# TYPE simplex_smp_subscribtion_clients_total gauge\n\
-      \simplex_smp_subscribtion_clients_total " <> mshow smpSubClientsCount <> "\n# smpSubClients\n\
+      \simplex_smp_subscribtion_clients_total " <> mshow (subClientsCount smpSubs) <> "\n# smp.subClientsCount\n\
       \\n\
       \# HELP simplex_smp_subscription_ntf_total Total notification subscripbtions (from ntf server)\n\
       \# TYPE simplex_smp_subscription_ntf_total gauge\n\
-      \simplex_smp_subscription_ntf_total " <> mshow ntfSubsCount <> "\n# ntfSubs\n\
+      \simplex_smp_subscription_ntf_total " <> mshow (subsCount ntfSubs) <> "\n# ntf.subsCount\n\
       \\n\
-      \# HELP simplex_smp_subscription_ntf_clients_total Total subscribed NTF servers, first counting method\n\
+      \# HELP simplex_smp_subscription_ntf_clients_total Total subscribed NTF servers\n\
       \# TYPE simplex_smp_subscription_ntf_clients_total gauge\n\
-      \simplex_smp_subscription_ntf_clients_total " <> mshow ntfSubClientsCount <> "\n# ntfSubClients\n\
+      \simplex_smp_subscription_ntf_clients_total " <> mshow (subClientsCount ntfSubs) <> "\n# ntf.subClientsCount\n\
       \\n\
       \# HELP simplex_smp_loaded_queues_queue_count Total loaded queues count (all queues for memory/journal storage)\n\
       \# TYPE simplex_smp_loaded_queues_queue_count gauge\n\

src/Simplex/Messaging/Server/QueueStore/Postgres.hs

@@ -25,6 +25,7 @@ module Simplex.Messaging.Server.QueueStore.Postgres
     foldQueueRecs,
     handleDuplicate,
     withLog_,
+    withDB',
   )
 where
 
@@ -138,7 +139,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
           db
           [sql|
             SELECT
-              (SELECT COUNT(1) FROM msg_queues WHERE deleted_at IS NULL) AS queue_count, 
+              (SELECT COUNT(1) FROM msg_queues WHERE deleted_at IS NULL) AS queue_count,
               (SELECT COUNT(1) FROM msg_queues WHERE deleted_at IS NULL AND notifier_id IS NOT NULL) AS notifier_count
           |]
       pure QueueCounts {queueCount, notifierCount}
@@ -221,7 +222,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
       _ -> throwE AUTH
 
   addQueueLinkData :: PostgresQueueStore q -> q -> LinkId -> QueueLinkData -> IO (Either ErrorType ())
-  addQueueLinkData st sq lnkId d = 
+  addQueueLinkData st sq lnkId d =
     withQueueRec sq "addQueueLinkData" $ \q -> case queueData q of
       Nothing ->
         addLink q $ \db -> DB.execute db qry (d :. (lnkId, rId))
@@ -335,7 +336,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
   unblockQueue st sq =
     setStatusDB "unblockQueue" st sq EntityActive $
       withLog "unblockQueue" st (`logUnblockQueue` recipientId sq)
-  
+
   updateQueueTime :: PostgresQueueStore q -> q -> RoundedSystemTime -> IO (Either ErrorType QueueRec)
   updateQueueTime st sq t =
     withQueueRec sq "updateQueueTime" $ \q@QueueRec {updatedAt} ->

src/Simplex/Messaging/TMap.hs

@@ -1,3 +1,5 @@
+{-# LANGUAGE TupleSections #-}
+
 module Simplex.Messaging.TMap
   ( TMap,
     emptyIO,
@@ -72,11 +74,11 @@ delete k m = modifyTVar' m $ M.delete k
 {-# INLINE delete #-}
 
 lookupInsert :: Ord k => k -> a -> TMap k a -> STM (Maybe a)
-lookupInsert k v m = stateTVar m $ \mv -> (M.lookup k mv, M.insert k v mv)
+lookupInsert k v m = stateTVar m $ M.alterF (,Just v) k
 {-# INLINE lookupInsert #-}
 
 lookupDelete :: Ord k => k -> TMap k a -> STM (Maybe a)
-lookupDelete k m = stateTVar m $ \mv -> (M.lookup k mv, M.delete k mv)
+lookupDelete k m = stateTVar m $ M.alterF (,Nothing) k
 {-# INLINE lookupDelete #-}
 
 adjust :: Ord k => (a -> a) -> k -> TMap k a -> STM ()

src/Simplex/Messaging/Transport.hs

@@ -61,7 +61,10 @@ module Simplex.Messaging.Transport
     Transport (..),
     TProxy (..),
     ATransport (..),
+    ASrvTransport,
     TransportPeer (..),
+    STransportPeer (..),
+    TransportPeerI (..),
     getServerVerifyKey,
 
     -- * TLS Transport
@@ -107,6 +110,7 @@ import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Default (def)
 import Data.Functor (($>))
+import Data.Kind (Type)
 import Data.Tuple (swap)
 import Data.Typeable (Typeable)
 import Data.Version (showVersion)
@@ -241,68 +245,75 @@ data TransportConfig = TransportConfig
     transportTimeout :: Maybe Int
   }
 
-class Typeable c => Transport c where
-  transport :: ATransport
-  transport = ATransport (TProxy @c)
+class Typeable c => Transport (c :: TransportPeer -> Type) where
+  transport :: forall p. ATransport p
+  transport = ATransport (TProxy @c @p)
 
-  transportName :: TProxy c -> String
+  transportName :: TProxy c p -> String
 
-  transportPeer :: c -> TransportPeer
+  transportConfig :: c p -> TransportConfig
 
-  transportConfig :: c -> TransportConfig
+  -- | Upgrade TLS context to connection
+  getTransportConnection :: TransportPeerI p => TransportConfig -> X.CertificateChain -> T.Context -> IO (c p)
 
-  -- | Upgrade server TLS context to connection (used in the server)
-  getServerConnection :: TransportConfig -> X.CertificateChain -> T.Context -> IO c
-
-  -- | Upgrade client TLS context to connection (used in the client)
-  getClientConnection :: TransportConfig -> X.CertificateChain -> T.Context -> IO c
-
-  getServerCerts :: c -> X.CertificateChain
+  -- | TLS certificate chain, server's in the client, client's in the server (empty chain)
+  getPeerCertChain :: c p -> X.CertificateChain
 
   -- | tls-unique channel binding per RFC5929
-  tlsUnique :: c -> SessionId
+  tlsUnique :: c p -> SessionId
 
   -- | ALPN value negotiated for the session
-  getSessionALPN :: c -> Maybe ALPN
+  getSessionALPN :: c p -> Maybe ALPN
 
   -- | Close connection
-  closeConnection :: c -> IO ()
+  closeConnection :: c p -> IO ()
 
   -- | Read fixed number of bytes from connection
-  cGet :: c -> Int -> IO ByteString
+  cGet :: c p -> Int -> IO ByteString
 
   -- | Write bytes to connection
-  cPut :: c -> ByteString -> IO ()
+  cPut :: c p -> ByteString -> IO ()
 
   -- | Receive ByteString from connection, allowing LF or CRLF termination.
-  getLn :: c -> IO ByteString
+  getLn :: c p -> IO ByteString
 
   -- | Send ByteString to connection terminating it with CRLF.
-  putLn :: c -> ByteString -> IO ()
+  putLn :: c p -> ByteString -> IO ()
   putLn c = cPut c . (<> "\r\n")
 
 data TransportPeer = TClient | TServer
   deriving (Eq, Show)
 
-data TProxy c = TProxy
+data STransportPeer (p :: TransportPeer) where
+  STClient :: STransportPeer 'TClient
+  STServer :: STransportPeer 'TServer
 
-data ATransport = forall c. Transport c => ATransport (TProxy c)
+class TransportPeerI p where sTransportPeer :: STransportPeer p
 
-getServerVerifyKey :: Transport c => c -> Either String C.APublicVerifyKey
+instance TransportPeerI 'TClient where sTransportPeer = STClient
+
+instance TransportPeerI 'TServer where sTransportPeer = STServer
+
+data TProxy (c :: TransportPeer -> Type) (p :: TransportPeer) = TProxy
+
+data ATransport p = forall c. Transport c => ATransport (TProxy c p)
+
+type ASrvTransport = ATransport 'TServer
+
+getServerVerifyKey :: Transport c => c 'TClient -> Either String C.APublicVerifyKey
 getServerVerifyKey c =
-  case getServerCerts c of
+  case getPeerCertChain c of
     X.CertificateChain (server : _ca) -> C.x509ToPublic (X.certPubKey . X.signedObject $ X.getSigned server, []) >>= C.pubKey
     _ -> Left "no certificate chain"
 
 -- * TLS Transport
 
-data TLS = TLS
+data TLS (p :: TransportPeer) = TLS
   { tlsContext :: T.Context,
-    tlsPeer :: TransportPeer,
     tlsUniq :: ByteString,
     tlsBuffer :: TBuffer,
     tlsALPN :: Maybe ALPN,
-    tlsServerCerts :: X.CertificateChain,
+    tlsPeerCert :: X.CertificateChain,
     tlsTransportConfig :: TransportConfig
   }
 
@@ -317,21 +328,22 @@ connectTLS host_ TransportConfig {logTLSErrors} params sock =
     logThrow e = putStrLn ("TLS error" <> host <> ": " <> show e) >> E.throwIO e
     host = maybe "" (\h -> " (" <> h <> ")") host_
 
-getTLS :: TransportPeer -> TransportConfig -> X.CertificateChain -> T.Context -> IO TLS
-getTLS tlsPeer cfg tlsServerCerts cxt = withTlsUnique tlsPeer cxt newTLS
+getTLS :: forall p. TransportPeerI p => TransportConfig -> X.CertificateChain -> T.Context -> IO (TLS p)
+getTLS cfg tlsPeerCert cxt = withTlsUnique @TLS @p cxt newTLS
   where
     newTLS tlsUniq = do
       tlsBuffer <- newTBuffer
       tlsALPN <- T.getNegotiatedProtocol cxt
-      pure TLS {tlsContext = cxt, tlsALPN, tlsTransportConfig = cfg, tlsServerCerts, tlsPeer, tlsUniq, tlsBuffer}
+      pure TLS {tlsContext = cxt, tlsALPN, tlsTransportConfig = cfg, tlsPeerCert, tlsUniq, tlsBuffer}
 
-withTlsUnique :: TransportPeer -> T.Context -> (ByteString -> IO c) -> IO c
-withTlsUnique peer cxt f =
-  cxtFinished peer cxt
+withTlsUnique :: forall c p. TransportPeerI p => T.Context -> (ByteString -> IO (c p)) -> IO (c p)
+withTlsUnique cxt f =
+  cxtFinished cxt
     >>= maybe (closeTLS cxt >> ioe_EOF) f
   where
-    cxtFinished TServer = T.getPeerFinished
-    cxtFinished TClient = T.getFinished
+    cxtFinished = case sTransportPeer @p of
+      STServer -> T.getPeerFinished
+      STClient -> T.getFinished
 
 closeTLS :: T.Context -> IO ()
 closeTLS ctx =
@@ -375,26 +387,31 @@ defaultSupportedParamsHTTPS =
 
 instance Transport TLS where
   transportName _ = "TLS"
-  transportPeer = tlsPeer
+  {-# INLINE transportName #-}
   transportConfig = tlsTransportConfig
-  getServerConnection = getTLS TServer
-  getClientConnection = getTLS TClient
-  getServerCerts = tlsServerCerts
+  {-# INLINE transportConfig #-}
+  getTransportConnection = getTLS
+  {-# INLINE getTransportConnection #-}
+  getPeerCertChain = tlsPeerCert
+  {-# INLINE getPeerCertChain #-}
   getSessionALPN = tlsALPN
+  {-# INLINE getSessionALPN #-}
   tlsUnique = tlsUniq
+  {-# INLINE tlsUnique #-}
   closeConnection tls = closeTLS $ tlsContext tls
+  {-# INLINE closeConnection #-}
 
   -- https://hackage.haskell.org/package/tls-1.6.0/docs/Network-TLS.html#v:recvData
   -- this function may return less than requested number of bytes
-  cGet :: TLS -> Int -> IO ByteString
+  cGet :: TLS p -> Int -> IO ByteString
   cGet TLS {tlsContext, tlsBuffer, tlsTransportConfig = TransportConfig {transportTimeout = t_}} n =
     getBuffered tlsBuffer n t_ (T.recvData tlsContext)
 
-  cPut :: TLS -> ByteString -> IO ()
+  cPut :: TLS p -> ByteString -> IO ()
   cPut TLS {tlsContext, tlsTransportConfig = TransportConfig {transportTimeout = t_}} =
     withTimedErr t_ . T.sendData tlsContext . LB.fromStrict
 
-  getLn :: TLS -> IO ByteString
+  getLn :: TLS p -> IO ByteString
   getLn TLS {tlsContext, tlsBuffer} = do
     getLnBuffered tlsBuffer (T.recvData tlsContext) `E.catches` [E.Handler handleTlsEOF, E.Handler handleEOF]
     where
@@ -407,7 +424,7 @@ instance Transport TLS where
 
 -- | The handle for SMP encrypted transport connection over Transport.
 data THandle v c p = THandle
-  { connection :: c,
+  { connection :: c p,
     params :: THandleParams v p
   }
 
@@ -587,13 +604,12 @@ tGetBlock THandle {connection = c, params = THandleParams {blockSize, encryptBlo
 -- | Server SMP transport handshake.
 --
 -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#appendix-a
-smpServerHandshake :: forall c. Transport c => C.APrivateSignKey -> c -> C.KeyPairX25519 -> C.KeyHash -> VersionRangeSMP -> ExceptT TransportError IO (THandleSMP c 'TServer)
-smpServerHandshake serverSignKey c (k, pk) kh smpVRange = do
+smpServerHandshake :: forall c. Transport c => X.CertificateChain -> C.APrivateSignKey -> c 'TServer -> C.KeyPairX25519 -> C.KeyHash -> VersionRangeSMP -> ExceptT TransportError IO (THandleSMP c 'TServer)
+smpServerHandshake srvCert srvSignKey c (k, pk) kh smpVRange = do
   let th@THandle {params = THandleParams {sessionId}} = smpTHandle c
-      sk = C.signX509 serverSignKey $ C.publicToX509 k
-      certChain = getServerCerts c
+      sk = C.signX509 srvSignKey $ C.publicToX509 k
       smpVersionRange = maybe legacyServerSMPRelayVRange (const smpVRange) $ getSessionALPN c
-  sendHandshake th $ ServerHandshake {sessionId, smpVersionRange, authPubKey = Just (certChain, sk)}
+  sendHandshake th $ ServerHandshake {sessionId, smpVersionRange, authPubKey = Just (srvCert, sk)}
   getHandshake th >>= \case
     ClientHandshake {smpVersion = v, keyHash, authPubKey = k', proxyServer}
       | keyHash /= kh ->
@@ -606,7 +622,7 @@ smpServerHandshake serverSignKey c (k, pk) kh smpVRange = do
 -- | Client SMP transport handshake.
 --
 -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#appendix-a
-smpClientHandshake :: forall c. Transport c => c -> Maybe C.KeyPairX25519 -> C.KeyHash -> VersionRangeSMP -> Bool -> ExceptT TransportError IO (THandleSMP c 'TClient)
+smpClientHandshake :: forall c. Transport c => c 'TClient -> Maybe C.KeyPairX25519 -> C.KeyHash -> VersionRangeSMP -> Bool -> ExceptT TransportError IO (THandleSMP c 'TClient)
 smpClientHandshake c ks_ keyHash@(C.KeyHash kh) vRange proxyServer = do
   let th@THandle {params = THandleParams {sessionId}} = smpTHandle c
   ServerHandshake {sessionId = sessId, smpVersionRange, authPubKey} <- getHandshake th
@@ -689,7 +705,7 @@ sendHandshake th = ExceptT . tPutBlock th . smpEncode
 getHandshake :: (Transport c, Encoding smp) => THandle v c p -> ExceptT TransportError IO smp
 getHandshake th = ExceptT $ (first (\_ -> TEHandshake PARSE) . A.parseOnly smpP =<<) <$> tGetBlock th
 
-smpTHandle :: Transport c => c -> THandleSMP c p
+smpTHandle :: Transport c => c p -> THandleSMP c p
 smpTHandle c = THandle {connection = c, params}
   where
     v = VersionSMP 0

src/Simplex/Messaging/Transport/Client.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE LambdaCase #-}
@@ -142,10 +143,10 @@ clientTransportConfig TransportClientConfig {logTLSErrors} =
   TransportConfig {logTLSErrors, transportTimeout = Nothing}
 
 -- | Connect to passed TCP host:port and pass handle to the client.
-runTransportClient :: Transport c => TransportClientConfig -> Maybe SocksCredentials -> TransportHost -> ServiceName -> Maybe C.KeyHash -> (c -> IO a) -> IO a
+runTransportClient :: Transport c => TransportClientConfig -> Maybe SocksCredentials -> TransportHost -> ServiceName -> Maybe C.KeyHash -> (c 'TClient -> IO a) -> IO a
 runTransportClient = runTLSTransportClient defaultSupportedParams Nothing
 
-runTLSTransportClient :: Transport c => T.Supported -> Maybe XS.CertificateStore -> TransportClientConfig -> Maybe SocksCredentials -> TransportHost -> ServiceName -> Maybe C.KeyHash -> (c -> IO a) -> IO a
+runTLSTransportClient :: Transport c => T.Supported -> Maybe XS.CertificateStore -> TransportClientConfig -> Maybe SocksCredentials -> TransportHost -> ServiceName -> Maybe C.KeyHash -> (c 'TClient -> IO a) -> IO a
 runTLSTransportClient tlsParams caStore_ cfg@TransportClientConfig {socksProxy, tcpKeepAlive, clientCredentials, alpn, useSNI} socksCreds host port keyHash client = do
   serverCert <- newEmptyTMVarIO
   let hostName = B.unpack $ strEncode host
@@ -165,7 +166,7 @@ runTLSTransportClient tlsParams caStore_ cfg@TransportClientConfig {socksProxy,
           logError "onServerCertificate didn't fire or failed to get cert chain"
           closeTLS tls >> error "onServerCertificate failed"
         Just c -> pure c
-    getClientConnection tCfg chain tls
+    getTransportConnection tCfg chain tls
   client c `E.finally` closeConnection c
   where
     hostAddr = \case

src/Simplex/Messaging/Transport/HTTP2.hs

@@ -22,10 +22,10 @@ import qualified System.TimeManager as TI
 defaultHTTP2BufferSize :: BufferSize
 defaultHTTP2BufferSize = 32768
 
-withHTTP2 :: BufferSize -> (Config -> IO a) -> IO () -> TLS -> IO a
+withHTTP2 :: BufferSize -> (Config -> IO a) -> IO () -> TLS p -> IO a
 withHTTP2 sz run fin c = E.bracket (allocHTTP2Config c sz) (\cfg -> freeSimpleConfig cfg `E.finally` fin) run
 
-allocHTTP2Config :: TLS -> BufferSize -> IO Config
+allocHTTP2Config :: TLS p -> BufferSize -> IO Config
 allocHTTP2Config c sz = do
   buf <- mallocBytes sz
   tm <- TI.initialize $ 30 * 1000000

src/Simplex/Messaging/Transport/HTTP2/Client.hs

@@ -1,9 +1,12 @@
+{-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
 
 module Simplex.Messaging.Transport.HTTP2.Client where
 
@@ -24,7 +27,7 @@ import qualified Network.TLS as T
 import Numeric.Natural (Natural)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Transport (ALPN, SessionId, TLS (tlsALPN), getServerCerts, getServerVerifyKey, tlsUniq)
+import Simplex.Messaging.Transport (ALPN, STransportPeer (..), SessionId, TLS (tlsALPN, tlsPeerCert, tlsUniq), TransportPeer (..), TransportPeerI (..), getServerVerifyKey)
 import Simplex.Messaging.Transport.Client (TransportClientConfig (..), TransportHost (..), defaultTcpConnectTimeout, runTLSTransportClient)
 import Simplex.Messaging.Transport.HTTP2
 import Simplex.Messaging.Util (eitherToMaybe)
@@ -97,13 +100,14 @@ getVerifiedHTTP2Client socksCreds host port keyHash caStore config disconnected
   where
     setup = runHTTP2Client (suportedTLSParams config) caStore (transportConfig config) (bufferSize config) socksCreds host port keyHash
 
-attachHTTP2Client :: HTTP2ClientConfig -> TransportHost -> ServiceName -> IO () -> Int -> TLS -> IO (Either HTTP2ClientError HTTP2Client)
+-- HTTP2 client can be run on both client and server TLS connections.
+attachHTTP2Client :: forall p. TransportPeerI p => HTTP2ClientConfig -> TransportHost -> ServiceName -> IO () -> Int -> TLS p -> IO (Either HTTP2ClientError HTTP2Client)
 attachHTTP2Client config host port disconnected bufferSize tls = getVerifiedHTTP2ClientWith config host port disconnected setup
   where
-    setup :: (TLS -> H.Client HTTP2Response) -> IO HTTP2Response
+    setup :: (TLS p -> H.Client HTTP2Response) -> IO HTTP2Response
     setup = runHTTP2ClientWith bufferSize host ($ tls)
 
-getVerifiedHTTP2ClientWith :: HTTP2ClientConfig -> TransportHost -> ServiceName -> IO () -> ((TLS -> H.Client HTTP2Response) -> IO HTTP2Response) -> IO (Either HTTP2ClientError HTTP2Client)
+getVerifiedHTTP2ClientWith :: forall p. TransportPeerI p => HTTP2ClientConfig -> TransportHost -> ServiceName -> IO () -> ((TLS p -> H.Client HTTP2Response) -> IO HTTP2Response) -> IO (Either HTTP2ClientError HTTP2Client)
 getVerifiedHTTP2ClientWith config host port disconnected setup =
   (mkHTTPS2Client >>= runClient)
     `E.catch` \(e :: IOException) -> pure . Left $ HCIOError e
@@ -124,15 +128,17 @@ getVerifiedHTTP2ClientWith config host port disconnected setup =
         Just (Left e) -> pure $ Left e
         Nothing -> cancel action $> Left HCNetworkError
 
-    client :: HClient -> TMVar (Either HTTP2ClientError HTTP2Client) -> TLS -> H.Client HTTP2Response
+    client :: HClient -> TMVar (Either HTTP2ClientError HTTP2Client) -> TLS p -> H.Client HTTP2Response
     client c cVar tls sendReq = do
       sessionTs <- getCurrentTime
       let c' =
             HTTP2Client
               { action = Nothing,
                 client_ = c,
-                serverKey = eitherToMaybe $ getServerVerifyKey tls,
-                serverCerts = getServerCerts tls,
+                serverKey = case sTransportPeer @p of
+                  STClient -> eitherToMaybe $ getServerVerifyKey tls
+                  STServer -> Nothing,
+                serverCerts = tlsPeerCert tls,
                 sendReq,
                 sessionTs,
                 sessionId = tlsUniq tls,
@@ -179,14 +185,15 @@ sendRequestDirect HTTP2Client {client_ = HClient {config, disconnected}, sendReq
 http2RequestTimeout :: HTTP2ClientConfig -> Maybe Int -> Int
 http2RequestTimeout HTTP2ClientConfig {connTimeout} = maybe connTimeout (connTimeout +)
 
-runHTTP2Client :: forall a. T.Supported -> Maybe XS.CertificateStore -> TransportClientConfig -> BufferSize -> Maybe SocksCredentials -> TransportHost -> ServiceName -> Maybe C.KeyHash -> (TLS -> H.Client a) -> IO a
+runHTTP2Client :: forall a. T.Supported -> Maybe XS.CertificateStore -> TransportClientConfig -> BufferSize -> Maybe SocksCredentials -> TransportHost -> ServiceName -> Maybe C.KeyHash -> (TLS 'TClient -> H.Client a) -> IO a
 runHTTP2Client tlsParams caStore tcConfig bufferSize socksCreds host port keyHash = runHTTP2ClientWith bufferSize host setup
   where
-    setup :: (TLS -> IO a) -> IO a
+    setup :: (TLS 'TClient -> IO a) -> IO a
     setup = runTLSTransportClient tlsParams caStore tcConfig socksCreds host port keyHash
 
-runHTTP2ClientWith :: forall a. BufferSize -> TransportHost -> ((TLS -> IO a) -> IO a) -> (TLS -> H.Client a) -> IO a
+-- HTTP2 client can be run on both client and server TLS connections.
+runHTTP2ClientWith :: forall a p. BufferSize -> TransportHost -> ((TLS p -> IO a) -> IO a) -> (TLS p -> H.Client a) -> IO a
 runHTTP2ClientWith bufferSize host setup client = setup $ \tls -> withHTTP2 bufferSize (run tls) (pure ()) tls
   where
-    run :: TLS -> H.Config -> IO a
+    run :: TLS p -> H.Config -> IO a
     run tls cfg = H.run (ClientConfig "https" (strEncode host) 20) cfg $ client tls

src/Simplex/Messaging/Transport/HTTP2/Server.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE DataKinds #-}
 {-# LANGUAGE NamedFieldPuns #-}
 
 module Simplex.Messaging.Transport.HTTP2.Server where
@@ -67,10 +68,11 @@ runHTTP2Server started port bufferSize srvSupported srvCreds alpn_ transportConf
   where
     setup = runTransportServer started port srvSupported srvCreds alpn_ transportConfig
 
-runHTTP2ServerWith :: BufferSize -> ((TLS -> IO ()) -> a) -> HTTP2ServerFunc -> a
+-- HTTP2 server can be run on both client and server TLS connections.
+runHTTP2ServerWith :: BufferSize -> ((TLS p -> IO ()) -> a) -> HTTP2ServerFunc -> a
 runHTTP2ServerWith = runHTTP2ServerWith_ Nothing (\_sessId -> pure ())
 
-runHTTP2ServerWith_ :: Maybe ExpirationConfig -> (SessionId -> IO ()) -> BufferSize -> ((TLS -> IO ()) -> a) -> HTTP2ServerFunc -> a
+runHTTP2ServerWith_ :: Maybe ExpirationConfig -> (SessionId -> IO ()) -> BufferSize -> ((TLS p -> IO ()) -> a) -> HTTP2ServerFunc -> a
 runHTTP2ServerWith_ expCfg_ clientFinished bufferSize setup http2Server = setup $ \tls -> do
   activeAt <- newTVarIO =<< getSystemTime
   tid_ <- mapM (forkIO . expireInactiveClient tls activeAt) expCfg_

src/Simplex/Messaging/Transport/Server.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
@@ -87,31 +88,31 @@ serverTransportConfig TransportServerConfig {logTLSErrors} =
 -- | Run transport server (plain TCP or WebSockets) on passed TCP port and signal when server started and stopped via passed TMVar.
 --
 -- All accepted connections are passed to the passed function.
-runTransportServer :: forall c. Transport c => TMVar Bool -> ServiceName -> T.Supported -> T.Credential -> Maybe [ALPN] -> TransportServerConfig -> (c -> IO ()) -> IO ()
+runTransportServer :: forall c. Transport c => TMVar Bool -> ServiceName -> T.Supported -> T.Credential -> Maybe [ALPN] -> TransportServerConfig -> (c 'TServer -> IO ()) -> IO ()
 runTransportServer started port srvSupported srvCreds alpn_ cfg server = do
   ss <- newSocketState
   runTransportServerState ss started port srvSupported srvCreds alpn_ cfg server
 
-runTransportServerState :: forall c . Transport c => SocketState -> TMVar Bool -> ServiceName -> T.Supported -> T.Credential -> Maybe [ALPN] -> TransportServerConfig -> (c -> IO ()) -> IO ()
+runTransportServerState :: forall c . Transport c => SocketState -> TMVar Bool -> ServiceName -> T.Supported -> T.Credential -> Maybe [ALPN] -> TransportServerConfig -> (c 'TServer -> IO ()) -> IO ()
 runTransportServerState ss started port srvSupported srvCreds alpn_ cfg server = runTransportServerState_ ss started port srvSupported (const srvCreds) alpn_ cfg (const server)
 
-runTransportServerState_ :: forall c . Transport c => SocketState -> TMVar Bool -> ServiceName -> T.Supported -> (Maybe HostName -> T.Credential) -> Maybe [ALPN] -> TransportServerConfig -> (Socket -> c -> IO ()) -> IO ()
-runTransportServerState_ ss started port = runTransportServerSocketState ss started (startTCPServer started Nothing port) (transportName (TProxy :: TProxy c))
+runTransportServerState_ :: forall c . Transport c => SocketState -> TMVar Bool -> ServiceName -> T.Supported -> (Maybe HostName -> T.Credential) -> Maybe [ALPN] -> TransportServerConfig -> (Socket -> c 'TServer -> IO ()) -> IO ()
+runTransportServerState_ ss started port = runTransportServerSocketState ss started (startTCPServer started Nothing port) (transportName (TProxy :: TProxy c 'TServer))
 
 -- | Run a transport server with provided connection setup and handler.
-runTransportServerSocket :: Transport a => TMVar Bool -> IO Socket -> String -> T.Credential -> T.ServerParams -> TransportServerConfig -> (a -> IO ()) -> IO ()
+runTransportServerSocket :: Transport c => TMVar Bool -> IO Socket -> String -> T.Credential -> T.ServerParams -> TransportServerConfig -> (c 'TServer -> IO ()) -> IO ()
 runTransportServerSocket started getSocket threadLabel srvCreds srvParams cfg server = do
   ss <- newSocketState
   runTransportServerSocketState_ ss started getSocket threadLabel (const srvCreds) srvParams cfg (const server)
 
-runTransportServerSocketState :: Transport a => SocketState -> TMVar Bool -> IO Socket -> String -> T.Supported -> (Maybe HostName -> T.Credential) -> Maybe [ALPN] -> TransportServerConfig -> (Socket -> a -> IO ()) -> IO ()
+runTransportServerSocketState :: Transport c => SocketState -> TMVar Bool -> IO Socket -> String -> T.Supported -> (Maybe HostName -> T.Credential) -> Maybe [ALPN] -> TransportServerConfig -> (Socket -> c 'TServer -> IO ()) -> IO ()
 runTransportServerSocketState ss started getSocket threadLabel srvSupported srvCreds alpn_ =
   runTransportServerSocketState_ ss started getSocket threadLabel srvCreds srvParams
   where
     srvParams = supportedTLSServerParams_ srvSupported srvCreds alpn_
 
 -- | Run a transport server with provided connection setup and handler.
-runTransportServerSocketState_ :: Transport a => SocketState -> TMVar Bool -> IO Socket -> String -> (Maybe HostName -> (X.CertificateChain, X.PrivKey)) -> T.ServerParams -> TransportServerConfig -> (Socket -> a -> IO ()) -> IO ()
+runTransportServerSocketState_ :: Transport c => SocketState -> TMVar Bool -> IO Socket -> String -> (Maybe HostName -> (X.CertificateChain, X.PrivKey)) -> T.ServerParams -> TransportServerConfig -> (Socket -> c 'TServer -> IO ()) -> IO ()
 runTransportServerSocketState_ ss started getSocket threadLabel srvCreds srvParams cfg server = do
   labelMyThread $ "transport server for " <> threadLabel
   runTCPServerSocket ss started getSocket $ \conn ->
@@ -121,7 +122,7 @@ runTransportServerSocketState_ ss started getSocket threadLabel srvCreds srvPara
     setup conn = timeout (tlsSetupTimeout cfg) $ do
       labelMyThread $ threadLabel <> "/setup"
       tls <- connectTLS Nothing tCfg srvParams conn
-      getServerConnection tCfg (fst $ srvCreds Nothing) tls
+      getTransportConnection tCfg (fst $ srvCreds Nothing) tls
 
 -- | Run TCP server without TLS
 runLocalTCPServer :: TMVar Bool -> ServiceName -> (Socket -> IO ()) -> IO ()

src/Simplex/Messaging/Transport/WebSockets.hs

@@ -1,6 +1,11 @@
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE DataKinds #-}
 {-# LANGUAGE InstanceSigs #-}
+{-# LANGUAGE KindSignatures #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
 
 module Simplex.Messaging.Transport.WebSockets (WS (..)) where
 
@@ -15,11 +20,12 @@ import Network.WebSockets.Stream (Stream)
 import qualified Network.WebSockets.Stream as S
 import Simplex.Messaging.Transport
   ( ALPN,
-    TProxy,
     Transport (..),
     TransportConfig (..),
     TransportError (..),
     TransportPeer (..),
+    STransportPeer (..),
+    TransportPeerI (..),
     closeTLS,
     smpBlockSize,
     withTlsUnique,
@@ -27,14 +33,13 @@ import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Buffer (trimCR)
 import System.IO.Error (isEOFError)
 
-data WS = WS
-  { wsPeer :: TransportPeer,
-    tlsUniq :: ByteString,
+data WS (p :: TransportPeer) = WS
+  { tlsUniq :: ByteString,
     wsALPN :: Maybe ALPN,
     wsStream :: Stream,
     wsConnection :: Connection,
     wsTransportConfig :: TransportConfig,
-    wsServerCerts :: X.CertificateChain
+    wsPeerCert :: X.CertificateChain
   }
 
 websocketsOpts :: ConnectionOptions
@@ -46,61 +51,50 @@ websocketsOpts =
     }
 
 instance Transport WS where
-  transportName :: TProxy WS -> String
   transportName _ = "WebSockets"
-
-  transportPeer :: WS -> TransportPeer
-  transportPeer = wsPeer
-
-  transportConfig :: WS -> TransportConfig
+  {-# INLINE transportName #-}
   transportConfig = wsTransportConfig
-
-  getServerConnection :: TransportConfig -> X.CertificateChain -> T.Context -> IO WS
-  getServerConnection = getWS TServer
-
-  getClientConnection :: TransportConfig -> X.CertificateChain -> T.Context -> IO WS
-  getClientConnection = getWS TClient
-
-  getServerCerts :: WS -> X.CertificateChain
-  getServerCerts = wsServerCerts
-
-  getSessionALPN :: WS -> Maybe ALPN
+  {-# INLINE transportConfig #-}
+  getTransportConnection = getWS
+  {-# INLINE getTransportConnection #-}
+  getPeerCertChain = wsPeerCert
+  {-# INLINE getPeerCertChain #-}
   getSessionALPN = wsALPN
-
-  tlsUnique :: WS -> ByteString
+  {-# INLINE getSessionALPN #-}
   tlsUnique = tlsUniq
-
-  closeConnection :: WS -> IO ()
+  {-# INLINE tlsUnique #-}
   closeConnection = S.close . wsStream
+  {-# INLINE closeConnection #-}
 
-  cGet :: WS -> Int -> IO ByteString
+  cGet :: WS p -> Int -> IO ByteString
   cGet c n = do
     s <- receiveData (wsConnection c)
     if B.length s == n
       then pure s
       else E.throwIO TEBadBlock
 
-  cPut :: WS -> ByteString -> IO ()
+  cPut :: WS p -> ByteString -> IO ()
   cPut = sendBinaryData . wsConnection
 
-  getLn :: WS -> IO ByteString
+  getLn :: WS p -> IO ByteString
   getLn c = do
     s <- trimCR <$> receiveData (wsConnection c)
     if B.null s || B.last s /= '\n'
       then E.throwIO TEBadBlock
       else pure $ B.init s
 
-getWS :: TransportPeer -> TransportConfig -> X.CertificateChain -> T.Context -> IO WS
-getWS wsPeer cfg wsServerCerts cxt = withTlsUnique wsPeer cxt connectWS
+getWS :: forall p. TransportPeerI p => TransportConfig -> X.CertificateChain -> T.Context -> IO (WS p)
+getWS cfg wsPeerCert cxt = withTlsUnique @WS @p cxt connectWS
   where
     connectWS tlsUniq = do
       s <- makeTLSContextStream cxt
-      wsConnection <- connectPeer wsPeer s
+      wsConnection <- connectPeer s
       wsALPN <- T.getNegotiatedProtocol cxt
-      pure $ WS {wsPeer, tlsUniq, wsALPN, wsStream = s, wsConnection, wsTransportConfig = cfg, wsServerCerts}
-    connectPeer :: TransportPeer -> Stream -> IO Connection
-    connectPeer TServer = acceptClientRequest
-    connectPeer TClient = sendClientRequest
+      pure $ WS {tlsUniq, wsALPN, wsStream = s, wsConnection, wsTransportConfig = cfg, wsPeerCert}
+    connectPeer :: Stream -> IO Connection
+    connectPeer = case sTransportPeer @p of
+      STServer -> acceptClientRequest
+      STClient -> sendClientRequest
     acceptClientRequest s = makePendingConnectionFromStream s websocketsOpts >>= acceptRequest
     sendClientRequest s = newClientConnection s "" "/" websocketsOpts []
 

src/Simplex/RemoteControl/Client.hs

@@ -49,7 +49,7 @@ import qualified Data.Text as T
 import Data.Time.Clock.System (getSystemTime)
 import Data.Tuple (swap)
 import Data.Word (Word16)
-import qualified Data.X509 as X509
+import qualified Data.X509 as X
 import Data.X509.Validation (Fingerprint (..), getFingerprint)
 import Network.Socket (PortNumber, SockAddr (..), hostAddressToTuple)
 import qualified Network.TLS as TLS
@@ -62,7 +62,7 @@ import Simplex.Messaging.Crypto.SNTRUP761
 import Simplex.Messaging.Crypto.SNTRUP761.Bindings
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String (StrEncoding (..))
-import Simplex.Messaging.Transport (TSbChainKeys (..), TLS (..), cGet, cPut)
+import Simplex.Messaging.Transport (TSbChainKeys (..), TLS (..), TransportPeer (..), cGet, cPut)
 import Simplex.Messaging.Transport.Buffer (peekBuffered)
 import Simplex.Messaging.Transport.Client (TransportClientConfig (..), TransportHost (..), defaultTransportClientConfig, runTransportClient)
 import Simplex.Messaging.Transport.Credentials (genCredentials, tlsCredentials)
@@ -101,7 +101,7 @@ data RCHClient_ = RCHClient_
     endSession :: TMVar ()
   }
 
-type RCHostConnection = (NonEmpty RCCtrlAddress, RCSignedInvitation, RCHostClient, RCStepTMVar (SessionCode, TLS, RCStepTMVar (RCHostSession, RCHostHello, RCHostPairing)))
+type RCHostConnection = (NonEmpty RCCtrlAddress, RCSignedInvitation, RCHostClient, RCStepTMVar (SessionCode, TLS 'TServer, RCStepTMVar (RCHostSession, RCHostHello, RCHostPairing)))
 
 connectRCHost :: TVar ChaChaDRG -> RCHostPairing -> J.Value -> Bool -> Maybe RCCtrlAddress -> Maybe Word16 -> ExceptT RCErrorType IO RCHostConnection
 connectRCHost drg pairing@RCHostPairing {caKey, caCert, idPrivKey, knownHost} ctrlAppInfo multicast rcAddrPrefs_ port_ = do
@@ -131,7 +131,7 @@ connectRCHost drg pairing@RCHostPairing {caKey, caCert, idPrivKey, knownHost} ct
       endSession <- newEmptyTMVarIO
       hostCAHash <- newEmptyTMVarIO
       pure RCHClient_ {startedPort, announcer, hostCAHash, endSession}
-    runClient :: RCHClient_ -> RCStepTMVar (SessionCode, TLS, RCStepTMVar (RCHostSession, RCHostHello, RCHostPairing)) -> RCHostKeys -> IO (Async ())
+    runClient :: RCHClient_ -> RCStepTMVar (SessionCode, TLS 'TServer, RCStepTMVar (RCHostSession, RCHostHello, RCHostPairing)) -> RCHostKeys -> IO (Async ())
     runClient RCHClient_ {startedPort, announcer, hostCAHash, endSession} r hostKeys = do
       tlsCreds <- genTLSCredentials drg caKey caCert
       startTLSServer port_ startedPort tlsCreds (tlsHooks r knownHost hostCAHash) $ \tls ->
@@ -157,7 +157,7 @@ connectRCHost drg pairing@RCHostPairing {caKey, caCert, idPrivKey, knownHost} ct
     tlsHooks r knownHost_ hostCAHash =
       def
         { TLS.onNewHandshake = \_ -> atomically $ isNothing <$> tryReadTMVar r,
-          TLS.onClientCertificate = \(X509.CertificateChain chain) ->
+          TLS.onClientCertificate = \(X.CertificateChain chain) ->
             case chain of
               [_leaf, ca] -> do
                 let kh = certFingerprint ca
@@ -190,16 +190,16 @@ connectRCHost drg pairing@RCHostPairing {caKey, caCert, idPrivKey, knownHost} ct
               }
       pure $ signInvitation (snd sessKeys) idPrivKey inv
 
-genTLSCredentials :: TVar ChaChaDRG -> C.APrivateSignKey -> C.SignedCertificate -> IO TLS.Credential
+genTLSCredentials :: TVar ChaChaDRG -> C.APrivateSignKey -> X.SignedCertificate -> IO TLS.Credential
 genTLSCredentials drg caKey caCert = do
   let caCreds = (C.signatureKeyPair caKey, caCert)
   leaf <- genCredentials drg (Just caCreds) (0, 24 * 999999) "localhost" -- session-signing cert
   pure . snd $ tlsCredentials (leaf :| [caCreds])
 
-certFingerprint :: X509.SignedCertificate -> C.KeyHash
+certFingerprint :: X.SignedCertificate -> C.KeyHash
 certFingerprint caCert = C.KeyHash fp
   where
-    Fingerprint fp = getFingerprint caCert X509.HashSHA256
+    Fingerprint fp = getFingerprint caCert X.HashSHA256
 
 cancelHostClient :: RCHostClient -> IO ()
 cancelHostClient RCHostClient {action, client_ = RCHClient_ {announcer, endSession}} = do
@@ -249,7 +249,7 @@ data RCCClient_ = RCCClient_
     endSession :: TMVar ()
   }
 
-type RCCtrlConnection = (RCCtrlClient, RCStepTMVar (SessionCode, TLS, RCStepTMVar (RCCtrlSession, RCCtrlPairing)))
+type RCCtrlConnection = (RCCtrlClient, RCStepTMVar (SessionCode, TLS 'TClient, RCStepTMVar (RCCtrlSession, RCCtrlPairing)))
 
 -- app should determine whether it is a new or known pairing based on CA fingerprint in the invitation
 connectRCCtrl :: TVar ChaChaDRG -> RCVerifiedInvitation -> Maybe RCCtrlPairing -> J.Value -> ExceptT RCErrorType IO RCCtrlConnection
@@ -280,7 +280,7 @@ connectRCCtrl_ drg pairing'@RCCtrlPairing {caKey, caCert} inv@RCInvitation {ca,
       confirmSession <- newEmptyTMVarIO
       endSession <- newEmptyTMVarIO
       pure RCCClient_ {confirmSession, endSession}
-    runClient :: RCCClient_ -> RCStepTMVar (SessionCode, TLS, RCStepTMVar (RCCtrlSession, RCCtrlPairing)) -> ExceptT RCErrorType IO ()
+    runClient :: RCCClient_ -> RCStepTMVar (SessionCode, TLS 'TClient, RCStepTMVar (RCCtrlSession, RCCtrlPairing)) -> ExceptT RCErrorType IO ()
     runClient RCCClient_ {confirmSession, endSession} r = do
       clientCredentials <- liftIO $ Just <$> genTLSCredentials drg caKey caCert
       let clientConfig = defaultTransportClientConfig {clientCredentials}
@@ -315,12 +315,12 @@ catchRCError = catchAllErrors $ \e -> case fromException e of
 putRCError :: ExceptT RCErrorType IO a -> TMVar (Either RCErrorType b) -> ExceptT RCErrorType IO a
 a `putRCError` r = a `catchRCError` \e -> atomically (tryPutTMVar r $ Left e) >> throwE e
 
-sendRCPacket :: Encoding a => TLS -> a -> ExceptT RCErrorType IO ()
+sendRCPacket :: Encoding a => TLS p -> a -> ExceptT RCErrorType IO ()
 sendRCPacket tls pkt = do
   b <- liftEitherWith (const RCEBlockSize) $ C.pad (smpEncode pkt) xrcpBlockSize
   liftIO $ cPut tls b
 
-receiveRCPacket :: Encoding a => TLS -> ExceptT RCErrorType IO a
+receiveRCPacket :: Encoding a => TLS p -> ExceptT RCErrorType IO a
 receiveRCPacket tls = do
   b <- liftIO $ cGet tls xrcpBlockSize
   when (B.length b /= xrcpBlockSize) $ throwE RCEBlockSize

src/Simplex/RemoteControl/Discovery.hs

@@ -23,7 +23,7 @@ import Network.Info (IPv4 (..), NetworkInterface (..), getNetworkInterfaces)
 import qualified Network.Socket as N
 import qualified Network.TLS as TLS
 import qualified Network.UDP as UDP
-import Simplex.Messaging.Transport (defaultSupportedParams)
+import Simplex.Messaging.Transport (TransportPeer (..), defaultSupportedParams)
 import qualified Simplex.Messaging.Transport as Transport
 import Simplex.Messaging.Transport.Client (TransportHost (..))
 import Simplex.Messaging.Transport.Server (defaultTransportServerConfig, runTransportServerSocket, startTCPServer)
@@ -68,7 +68,7 @@ preferAddress RCCtrlAddress {address, interface} addrs =
     matchAddr RCCtrlAddress {address = a} = a == address
     matchIface RCCtrlAddress {interface = i} = i == interface
 
-startTLSServer :: Maybe Word16 -> TMVar (Maybe N.PortNumber) -> TLS.Credential -> TLS.ServerHooks -> (Transport.TLS -> IO ()) -> IO (Async ())
+startTLSServer :: Maybe Word16 -> TMVar (Maybe N.PortNumber) -> TLS.Credential -> TLS.ServerHooks -> (Transport.TLS 'TServer -> IO ()) -> IO (Async ())
 startTLSServer port_ startedOnPort credentials hooks server = async . liftIO $ do
   started <- newEmptyTMVarIO
   bracketOnError (startTCPServer started Nothing $ maybe "0" show port_) (\_e -> setPort Nothing) $ \socket ->

src/Simplex/RemoteControl/Types.hs

@@ -18,12 +18,13 @@ import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Text.Encoding (encodeUtf8)
 import Data.Word (Word16)
+import qualified Data.X509 as X
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.SNTRUP761.Bindings
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (defaultJSON, dropPrefix, sumTypeJSON)
-import Simplex.Messaging.Transport (TLS, TSbChainKeys)
+import Simplex.Messaging.Transport (TLS, TSbChainKeys, TransportPeer (..))
 import Simplex.Messaging.Transport.Client (TransportHost)
 import Simplex.Messaging.Util (safeDecodeUtf8)
 import Simplex.Messaging.Version (VersionRange, VersionScope, mkVersionRange)
@@ -140,7 +141,7 @@ $(JQ.deriveJSON defaultJSON {J.nullaryToObject = True} ''RCCtrlHello)
 -- | Long-term part of controller (desktop) connection to host (mobile)
 data RCHostPairing = RCHostPairing
   { caKey :: C.APrivateSignKey,
-    caCert :: C.SignedCertificate,
+    caCert :: X.SignedCertificate,
     idPrivKey :: C.PrivateKeyEd25519,
     knownHost :: Maybe KnownHostPairing
   }
@@ -159,7 +160,7 @@ data RCCtrlAddress = RCCtrlAddress
 -- | Long-term part of host (mobile) connection to controller (desktop)
 data RCCtrlPairing = RCCtrlPairing
   { caKey :: C.APrivateSignKey,
-    caCert :: C.SignedCertificate,
+    caCert :: X.SignedCertificate,
     ctrlFingerprint :: C.KeyHash, -- long-term identity of connected remote controller
     idPubKey :: C.PublicKeyEd25519,
     dhPrivKey :: C.PrivateKeyX25519,
@@ -173,7 +174,7 @@ data RCHostKeys = RCHostKeys
 
 -- Connected session with Host
 data RCHostSession = RCHostSession
-  { tls :: TLS,
+  { tls :: TLS 'TServer,
     sessionKeys :: HostSessKeys
   }
 
@@ -186,7 +187,7 @@ data HostSessKeys = HostSessKeys
 -- Host: RCCtrlPairing + RCInvitation => (RCCtrlSession, RCCtrlPairing)
 
 data RCCtrlSession = RCCtrlSession
-  { tls :: TLS,
+  { tls :: TLS 'TClient,
     sessionKeys :: CtrlSessKeys
   }
 

tests/AgentTests.hs

@@ -15,8 +15,8 @@ import AgentTests.MigrationTests (migrationTests)
 import AgentTests.ServerChoice (serverChoiceTests)
 import AgentTests.ShortLinkTests (shortLinkTests)
 import Simplex.Messaging.Server.Env.STM (AStoreType (..))
-import Simplex.Messaging.Transport (ATransport (..))
-import Test.Hspec
+import Simplex.Messaging.Transport (ASrvTransport)
+import Test.Hspec hiding (fit, it)
 
 #if defined(dbPostgres)
 import Fixtures
@@ -38,7 +38,7 @@ agentCoreTests = do
   describe "Double ratchet tests" doubleRatchetTests
   describe "Short link tests" shortLinkTests
 
-agentTests :: (ATransport, AStoreType) -> Spec
+agentTests :: (ASrvTransport, AStoreType) -> Spec
 agentTests ps = do
 #if defined(dbPostgres)
   after_ (dropAllSchemasExceptSystem testDBConnectInfo) $ do
@@ -47,7 +47,7 @@ agentTests ps = do
 #endif
     describe "Functional API" $ functionalAPITests ps
     describe "Chosen servers" serverChoiceTests
-#if defined(dbServerPostgres)    
+#if defined(dbServerPostgres)
     around_ (postgressBracket ntfTestServerDBConnectInfo) $
       describe "Notification tests" $ notificationTests ps
 #endif

tests/AgentTests/ConnectionRequestTests.hs

@@ -28,7 +28,8 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (EntityId (..), ProtocolServer (..), QueueMode (..), currentSMPClientVersion, supportedSMPClientVRange, pattern VersionSMPC)
 import Simplex.Messaging.ServiceScheme (ServiceScheme (..))
 import Simplex.Messaging.Version
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
+import Util
 
 srv :: SMPServer
 srv = SMPServer "smp.simplex.im,jjbyvoemxysm7qxap7m5d5m35jzv5qq6gnlv7s4rsn7tdwwmuqciwpid.onion" "5223" (C.KeyHash "\215m\248\251")
@@ -288,7 +289,7 @@ connectionRequestTests =
       smpEncodingTest queueV1NoPort
       smpEncodingTest connectionRequest
       -- smpEncodingTest connectionRequestNoQM -- this fails, because of queue mode patch
-      smpEncodingTest connectionRequestContact  -- this passes because of queue mode patch in ConnReqUriData encoding
+      smpEncodingTest connectionRequestContact -- this passes because of queue mode patch in ConnReqUriData encoding
       smpEncodingTest connectionRequest1
       smpEncodingTest connectionRequest2queues
       smpEncodingTest connectionRequestNew
@@ -334,12 +335,12 @@ connectionRequestTests =
       restoreShortLink [srv] (contact srv2 (LinkKey "0123456789abcdef0123456789abcdef"))
         `shouldBe` contact srv2 (LinkKey "0123456789abcdef0123456789abcdef")
       Right (lnk :: ConnShortLink 'CMContact) <- pure $ strDecode "https://localhost/a#4AkRDmhf64tdRlN406g8lJRg5OCmhD6ynIhi6glOcCM?p=7001&c=LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI"
-      Right (lnk' :: ConnShortLink 'CMContact) <- pure $ strDecode  "https://localhost/a#4AkRDmhf64tdRlN406g8lJRg5OCmhD6ynIhi6glOcCM"
+      Right (lnk' :: ConnShortLink 'CMContact) <- pure $ strDecode "https://localhost/a#4AkRDmhf64tdRlN406g8lJRg5OCmhD6ynIhi6glOcCM"
       let presetSrv :: SMPServer = "smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:7001"
       shortenShortLink [presetSrv] lnk `shouldBe` lnk'
       restoreShortLink [presetSrv] lnk' `shouldBe` lnk
       Right (inv :: ConnShortLink 'CMInvitation) <- pure $ strDecode "https://localhost/i#tnUaHYp8saREmyEHR93SBpl8ySHBchOt/LJ1ZQUzxH9Udb0jw5wmJACv5o6oe8e7BsX_hUCUMTSY?p=7001&c=LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI"
-      Right (inv' :: ConnShortLink 'CMInvitation) <- pure $ strDecode  "https://localhost/i#tnUaHYp8saREmyEHR93SBpl8ySHBchOt/LJ1ZQUzxH9Udb0jw5wmJACv5o6oe8e7BsX_hUCUMTSY"
+      Right (inv' :: ConnShortLink 'CMInvitation) <- pure $ strDecode "https://localhost/i#tnUaHYp8saREmyEHR93SBpl8ySHBchOt/LJ1ZQUzxH9Udb0jw5wmJACv5o6oe8e7BsX_hUCUMTSY"
       shortenShortLink [presetSrv] inv `shouldBe` inv'
       restoreShortLink [presetSrv] inv' `shouldBe` inv
   where

tests/AgentTests/DoubleRatchetTests.hs

@@ -26,13 +26,14 @@ import qualified Data.Map.Strict as M
 import Data.Type.Equality
 import Simplex.Messaging.Crypto (Algorithm (..), AlgorithmI, CryptoError, DhAlgorithm)
 import qualified Simplex.Messaging.Crypto as C
-import Simplex.Messaging.Crypto.SNTRUP761.Bindings
 import Simplex.Messaging.Crypto.Ratchet
+import Simplex.Messaging.Crypto.SNTRUP761.Bindings
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Util ((<$$>))
 import Simplex.Messaging.Version
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
+import Util
 
 doubleRatchetTests :: Spec
 doubleRatchetTests = do
@@ -82,7 +83,6 @@ runMessageTests initRatchets_ agreeRatchetKEMs = do
       withRatchets_ @X25519 initRatchets_ test
       withRatchets_ @X448 initRatchets_ test
 
-
 testAlgs :: (forall a. (AlgorithmI a, DhAlgorithm a) => C.SAlgorithm a -> IO ()) -> IO ()
 testAlgs test = test C.SX25519 >> test C.SX448
 

tests/AgentTests/EqInstances.hs

@@ -22,8 +22,6 @@ deriving instance Eq (StoredRcvQueue q)
 
 deriving instance Eq (StoredSndQueue q)
 
-deriving instance Eq (DBQueueId q)
-
 deriving instance Eq ClientNtfCreds
 
 deriving instance Eq ShortLinkCreds

tests/AgentTests/FunctionalAPITests.hs

@@ -71,18 +71,20 @@ import Data.List.NonEmpty (NonEmpty)
 import qualified Data.Map as M
 import Data.Maybe (isJust, isNothing)
 import qualified Data.Set as S
+import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1)
+import qualified Data.Text.IO as T
 import Data.Time.Clock (diffUTCTime, getCurrentTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Data.Type.Equality (testEquality, (:~:) (Refl))
 import Data.Word (Word16)
 import GHC.Stack (withFrozenCallStack)
 import SMPAgentClient
-import SMPClient (cfgJ2QS, cfgMS, prevRange, prevVersion, proxyCfgJ2QS, proxyCfgMS, testPort, testPort2, testStoreLogFile, withSmpServer, withSmpServers2, withSmpServerConfigOn, withSmpServerProxy, withSmpServersProxy2, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
+import SMPClient
 import Simplex.Messaging.Agent hiding (createConnection, joinConnection, sendMessage)
 import qualified Simplex.Messaging.Agent as A
 import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..), ServerQueueInfo (..), UserNetworkInfo (..), UserNetworkType (..), waitForUserNetwork)
-import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), InitialAgentServers (..), createAgentStore)
+import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), Env (..), InitialAgentServers (..), createAgentStore)
 import Simplex.Messaging.Agent.Protocol hiding (CON, CONF, INFO, REQ, SENT)
 import qualified Simplex.Messaging.Agent.Protocol as A
 import Simplex.Messaging.Agent.Store.Common (DBStore (..), withTransaction)
@@ -102,19 +104,28 @@ import Simplex.Messaging.Server.Env.STM (AServerStoreCfg (..), AStoreType (..),
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.MsgStore.Types (SMSType (..), SQSType (..))
 import Simplex.Messaging.Server.QueueStore.QueueInfo
-import Simplex.Messaging.Transport (ATransport (..), SMPVersion, VersionSMP, authCmdsSMPVersion, currentServerSMPRelayVersion, minClientSMPRelayVersion, minServerSMPRelayVersion, sendingProxySMPVersion, sndAuthKeySMPVersion, supportedSMPHandshakes, supportedServerSMPRelayVRange)
+import Simplex.Messaging.Transport (ASrvTransport, SMPVersion, VersionSMP, authCmdsSMPVersion, currentServerSMPRelayVersion, minClientSMPRelayVersion, minServerSMPRelayVersion, sendingProxySMPVersion, sndAuthKeySMPVersion, supportedSMPHandshakes, supportedServerSMPRelayVRange)
 import Simplex.Messaging.Util (bshow, diffToMicroseconds)
 import Simplex.Messaging.Version (VersionRange (..))
 import qualified Simplex.Messaging.Version as V
 import Simplex.Messaging.Version.Internal (Version (..))
 import System.Directory (copyFile, renameFile)
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
 import UnliftIO
 import Util
 import XFTPClient (testXFTPServer)
 #if defined(dbPostgres)
 import Fixtures
 #endif
+#if defined(dbServerPostgres)
+import qualified Database.PostgreSQL.Simple as PSQL
+import Simplex.Messaging.Agent.Store (Connection (..), StoredRcvQueue (..), SomeConn (..))
+import Simplex.Messaging.Agent.Store.AgentStore (getConn)
+import Simplex.Messaging.Server.MsgStore.Journal (JournalQueue)
+import Simplex.Messaging.Server.MsgStore.Types (QSType (..))
+import Simplex.Messaging.Server.QueueStore.Postgres
+import Simplex.Messaging.Server.QueueStore.Types (QueueStoreClass (..))
+#endif
 
 type AEntityTransmission e = (ACorrId, ConnId, AEvent e)
 
@@ -267,7 +278,7 @@ sendMessage c connId msgFlags msgBody = do
   liftIO $ pqEnc `shouldBe` PQEncOn
   pure msgId
 
-functionalAPITests :: (ATransport, AStoreType) -> Spec
+functionalAPITests :: (ASrvTransport, AStoreType) -> Spec
 functionalAPITests ps = do
   describe "Establishing duplex connection" $ do
     testMatrix2 ps runAgentClientTest
@@ -320,6 +331,7 @@ functionalAPITests ps = do
       it "should get 1-time link data after restart" $ testInviationShortLinkRestart ps
       it "should connect via contact short link after restart" $ testContactShortLinkRestart ps
       it "should connect via added contact short link after restart" $ testAddContactShortLinkRestart ps
+    it "should create and get short links with the old contact queues" $ testOldContactQueueShortLink ps
   describe "Message delivery" $ do
     describe "update connection agent version on received messages" $ do
       it "should increase if compatible, shouldn'ps decrease" $
@@ -486,7 +498,7 @@ functionalAPITests ps = do
     it "server should respond with queue and subscription information" $
       withSmpServer ps testServerQueueInfo
 
-testBasicAuth :: (ATransport, AStoreType) -> Bool -> (Maybe BasicAuth, VersionSMP) -> (Maybe BasicAuth, VersionSMP) -> (Maybe BasicAuth, VersionSMP) -> SndQueueSecured -> AgentMsgId -> IO Int
+testBasicAuth :: (ASrvTransport, AStoreType) -> Bool -> (Maybe BasicAuth, VersionSMP) -> (Maybe BasicAuth, VersionSMP) -> (Maybe BasicAuth, VersionSMP) -> SndQueueSecured -> AgentMsgId -> IO Int
 testBasicAuth (t, msType) allowNewQueues srv@(srvAuth, srvVersion) clnt1 clnt2 sqSecured baseId = do
   let testCfg = (cfgMS msType) {allowNewQueues, newQueueBasicAuth = srvAuth, smpServerVRange = V.mkVersionRange minServerSMPRelayVersion srvVersion}
       canCreate1 = canCreateQueue allowNewQueues srv clnt1
@@ -503,7 +515,7 @@ canCreateQueue :: Bool -> (Maybe BasicAuth, VersionSMP) -> (Maybe BasicAuth, Ver
 canCreateQueue allowNew (srvAuth, _) (clntAuth, _) =
   allowNew && (isNothing srvAuth || srvAuth == clntAuth)
 
-testMatrix2 :: HasCallStack => (ATransport, AStoreType) -> (PQSupport -> SndQueueSecured -> Bool -> AgentClient -> AgentClient -> AgentMsgId -> IO ()) -> Spec
+testMatrix2 :: HasCallStack => (ASrvTransport, AStoreType) -> (PQSupport -> SndQueueSecured -> Bool -> AgentClient -> AgentClient -> AgentMsgId -> IO ()) -> Spec
 testMatrix2 ps runTest = do
   it "current, via proxy" $ withSmpServerProxy ps $ runTestCfgServers2 agentCfg agentCfg initAgentServersProxy 1 $ runTest PQSupportOn True True
   it "v8, via proxy" $ withSmpServerProxy ps $ runTestCfgServers2 agentProxyCfgV8 agentProxyCfgV8 initAgentServersProxy 3 $ runTest PQSupportOn False True
@@ -512,7 +524,7 @@ testMatrix2 ps runTest = do
   it "prev to current" $ withSmpServer ps $ runTestCfg2 agentCfgVPrev agentCfg 1 $ runTest PQSupportOff False False
   it "current to prev" $ withSmpServer ps $ runTestCfg2 agentCfg agentCfgVPrev 1 $ runTest PQSupportOff False False
 
-testMatrix2Stress :: HasCallStack => (ATransport, AStoreType) -> (PQSupport -> SndQueueSecured -> Bool -> AgentClient -> AgentClient -> AgentMsgId -> IO ()) -> Spec
+testMatrix2Stress :: HasCallStack => (ASrvTransport, AStoreType) -> (PQSupport -> SndQueueSecured -> Bool -> AgentClient -> AgentClient -> AgentMsgId -> IO ()) -> Spec
 testMatrix2Stress ps runTest = do
   it "current, via proxy" $ withSmpServerProxy ps $ runTestCfgServers2 aCfg aCfg initAgentServersProxy 1 $ runTest PQSupportOn True True
   it "v8, via proxy" $ withSmpServerProxy ps $ runTestCfgServers2 aProxyCfgV8 aProxyCfgV8 initAgentServersProxy 1 $ runTest PQSupportOn False True
@@ -525,14 +537,14 @@ testMatrix2Stress ps runTest = do
     aProxyCfgV8 = agentProxyCfgV8 {messageRetryInterval = fastMessageRetryInterval}
     aCfgVPrev = agentCfgVPrev {messageRetryInterval = fastMessageRetryInterval}
 
-testBasicMatrix2 :: HasCallStack => (ATransport, AStoreType) -> (SndQueueSecured -> AgentClient -> AgentClient -> AgentMsgId -> IO ()) -> Spec
+testBasicMatrix2 :: HasCallStack => (ASrvTransport, AStoreType) -> (SndQueueSecured -> AgentClient -> AgentClient -> AgentMsgId -> IO ()) -> Spec
 testBasicMatrix2 ps runTest = do
   it "current" $ withSmpServer ps $ runTestCfg2 agentCfg agentCfg 1 $ runTest True
   it "prev" $ withSmpServer ps $ runTestCfg2 agentCfgVPrevPQ agentCfgVPrevPQ 1 $ runTest False
   it "prev to current" $ withSmpServer ps $ runTestCfg2 agentCfgVPrevPQ agentCfg 1 $ runTest False
   it "current to prev" $ withSmpServer ps $ runTestCfg2 agentCfg agentCfgVPrevPQ 1 $ runTest False
 
-testRatchetMatrix2 :: HasCallStack => (ATransport, AStoreType) -> (PQSupport -> SndQueueSecured -> Bool -> AgentClient -> AgentClient -> AgentMsgId -> IO ()) -> Spec
+testRatchetMatrix2 :: HasCallStack => (ASrvTransport, AStoreType) -> (PQSupport -> SndQueueSecured -> Bool -> AgentClient -> AgentClient -> AgentMsgId -> IO ()) -> Spec
 testRatchetMatrix2 ps runTest = do
   it "current, via proxy" $ withSmpServerProxy ps $ runTestCfgServers2 agentCfg agentCfg initAgentServersProxy 1 $ runTest PQSupportOn True True
   it "v8, via proxy" $ withSmpServerProxy ps $ runTestCfgServers2 agentProxyCfgV8 agentProxyCfgV8 initAgentServersProxy 3 $ runTest PQSupportOn False True
@@ -541,17 +553,17 @@ testRatchetMatrix2 ps runTest = do
   it "ratchets prev to current" $ withSmpServer ps $ runTestCfg2 agentCfgRatchetVPrev agentCfg 1 $ runTest PQSupportOff True False
   it "ratchets current to prev" $ withSmpServer ps $ runTestCfg2 agentCfg agentCfgRatchetVPrev 1 $ runTest PQSupportOff True False
 
-testServerMatrix2 :: HasCallStack => (ATransport, AStoreType) -> (InitialAgentServers -> IO ()) -> Spec
+testServerMatrix2 :: HasCallStack => (ASrvTransport, AStoreType) -> (InitialAgentServers -> IO ()) -> Spec
 testServerMatrix2 ps runTest = do
   it "1 server" $ withSmpServer ps $ runTest initAgentServers
   it "2 servers" $ withSmpServers2 ps $ runTest initAgentServers2
 
-testProxyMatrix :: HasCallStack => (ATransport, AStoreType) -> (Bool -> AgentClient -> AgentClient -> IO ()) -> Spec
+testProxyMatrix :: HasCallStack => (ASrvTransport, AStoreType) -> (Bool -> AgentClient -> AgentClient -> IO ()) -> Spec
 testProxyMatrix ps runTest = do
   it "2 servers, directly" $ withSmpServers2 ps $ withAgentClientsServers2 (agentCfg, initAgentServers) (agentCfg, initAgentServers2) $ runTest False
   it "2 servers, via proxy" $ withSmpServersProxy2 ps $ withAgentClientsServers2 (agentCfg, initAgentServersProxy) (agentCfg, initAgentServersProxy2) $ runTest True
 
-testProxyMatrixWithPrev :: HasCallStack => (ATransport, AStoreType) -> (Bool -> Bool -> AgentClient -> AgentClient -> IO ()) -> Spec
+testProxyMatrixWithPrev :: HasCallStack => (ASrvTransport, AStoreType) -> (Bool -> Bool -> AgentClient -> AgentClient -> IO ()) -> Spec
 testProxyMatrixWithPrev ps@(t, msType@(ASType qs _ms)) runTest = do
   it "2 servers, directly, curr clients, prev servers" $ withSmpServers2Prev $ withAgentClientsServers2 (agentCfg, initAgentServers) (agentCfg, initAgentServers2) $ runTest False True
   it "2 servers, via proxy, curr clients, prev servers" $ withSmpServersProxy2Prev $ withAgentClientsServers2 (agentCfg, initAgentServersProxy) (agentCfg, initAgentServersProxy2) $ runTest True True
@@ -564,13 +576,13 @@ testProxyMatrixWithPrev ps@(t, msType@(ASType qs _ms)) runTest = do
     withServers2 cfg1 cfg2 a =
       withSmpServerConfigOn t cfg1 testPort $ \_ -> withSmpServerConfigOn t cfg2 testPort2 $ \_ -> a
 
-testPQMatrix2 :: HasCallStack => (ATransport, AStoreType) -> (HasCallStack => (AgentClient, InitialKeys) -> (AgentClient, PQSupport) -> AgentMsgId -> IO ()) -> Spec
+testPQMatrix2 :: HasCallStack => (ASrvTransport, AStoreType) -> (HasCallStack => (AgentClient, InitialKeys) -> (AgentClient, PQSupport) -> AgentMsgId -> IO ()) -> Spec
 testPQMatrix2 = pqMatrix2_ True
 
-testPQMatrix2NoInv :: HasCallStack => (ATransport, AStoreType) -> (HasCallStack => (AgentClient, InitialKeys) -> (AgentClient, PQSupport) -> AgentMsgId -> IO ()) -> Spec
+testPQMatrix2NoInv :: HasCallStack => (ASrvTransport, AStoreType) -> (HasCallStack => (AgentClient, InitialKeys) -> (AgentClient, PQSupport) -> AgentMsgId -> IO ()) -> Spec
 testPQMatrix2NoInv = pqMatrix2_ False
 
-pqMatrix2_ :: HasCallStack => Bool -> (ATransport, AStoreType) -> (HasCallStack => (AgentClient, InitialKeys) -> (AgentClient, PQSupport) -> AgentMsgId -> IO ()) -> Spec
+pqMatrix2_ :: HasCallStack => Bool -> (ASrvTransport, AStoreType) -> (HasCallStack => (AgentClient, InitialKeys) -> (AgentClient, PQSupport) -> AgentMsgId -> IO ()) -> Spec
 pqMatrix2_ pqInv ps test = do
   it "dh/dh handshake" $ runTest $ \a b -> test (a, IKPQOff) (b, PQSupportOff)
   it "dh/pq handshake" $ runTest $ \a b -> test (a, IKPQOff) (b, PQSupportOn)
@@ -584,7 +596,7 @@ pqMatrix2_ pqInv ps test = do
 
 testPQMatrix3 ::
   HasCallStack =>
-  (ATransport, AStoreType) ->
+  (ASrvTransport, AStoreType) ->
   (HasCallStack => (AgentClient, InitialKeys) -> (AgentClient, PQSupport) -> (AgentClient, PQSupport) -> AgentMsgId -> IO ()) ->
   Spec
 testPQMatrix3 ps test = do
@@ -1047,7 +1059,7 @@ testAsyncBothOffline = do
     liftIO $ disposeAgentClient alice'
     liftIO $ disposeAgentClient bob'
 
-testAsyncServerOffline :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testAsyncServerOffline :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testAsyncServerOffline ps = withAgentClients2 $ \alice bob -> do
   -- create connection and shutdown the server
   (bobId, cReq) <- withSmpServerStoreLogOn ps testPort $ \_ ->
@@ -1063,6 +1075,7 @@ testAsyncServerOffline ps = withAgentClients2 $ \alice bob -> do
     liftIO $ do
       srv1 `shouldBe` testSMPServer
       conns1 `shouldBe` [bobId]
+    liftIO $ threadDelay 250000
     (aliceId, sqSecured) <- joinConnection bob 1 True cReq "bob's connInfo" SMSubscribe
     liftIO $ sqSecured `shouldBe` True
     ("", _, CONF confId _ "bob's connInfo") <- get alice
@@ -1072,7 +1085,7 @@ testAsyncServerOffline ps = withAgentClients2 $ \alice bob -> do
     get bob ##> ("", aliceId, CON)
     exchangeGreetings alice bobId bob aliceId
 
-testAllowConnectionClientRestart :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testAllowConnectionClientRestart :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testAllowConnectionClientRestart ps@(t, ASType qsType _) = do
   let initAgentServersSrv2 = initAgentServers {smp = userServers [testSMPServer2]}
   alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
@@ -1200,13 +1213,13 @@ testContactShortLink viaProxy a b =
       exchangeGreetingsViaProxy viaProxy a bId b aId
     -- update user data
     let updatedData = "updated user data"
-    shortLink' <- runRight $ setContactShortLink a contactId updatedData
+    shortLink' <- runRight $ setContactShortLink a contactId updatedData Nothing
     shortLink' `shouldBe` shortLink
     (connReq4, updatedConnData') <- runRight $ getConnShortLink c 1 shortLink
     connReq4 `shouldBe` connReq
     linkUserData updatedConnData' `shouldBe` updatedData
     -- one more time
-    shortLink2 <- runRight $ setContactShortLink a contactId updatedData
+    shortLink2 <- runRight $ setContactShortLink a contactId updatedData Nothing
     shortLink2 `shouldBe` shortLink
     -- delete short link
     runRight_ $ deleteContactShortLink a contactId
@@ -1219,7 +1232,7 @@ testAddContactShortLink viaProxy a b =
     (contactId, CCLink connReq0 Nothing) <- runRight $ A.createConnection a 1 True SCMContact Nothing Nothing CR.IKPQOn SMSubscribe
     Right connReq <- pure $ smpDecode (smpEncode connReq0) --
     let userData = "some user data"
-    shortLink <- runRight $ setContactShortLink a contactId userData
+    shortLink <- runRight $ setContactShortLink a contactId userData Nothing
     (connReq', connData') <- runRight $ getConnShortLink b 1 shortLink
     strDecode (strEncode shortLink) `shouldBe` Right shortLink
     connReq' `shouldBe` connReq
@@ -1247,13 +1260,13 @@ testAddContactShortLink viaProxy a b =
       exchangeGreetingsViaProxy viaProxy a bId b aId
     -- update user data
     let updatedData = "updated user data"
-    shortLink' <- runRight $ setContactShortLink a contactId updatedData
+    shortLink' <- runRight $ setContactShortLink a contactId updatedData Nothing
     shortLink' `shouldBe` shortLink
     (connReq4, updatedConnData') <- runRight $ getConnShortLink c 1 shortLink
     connReq4 `shouldBe` connReq
     linkUserData updatedConnData' `shouldBe` updatedData
 
-testInviationShortLinkRestart :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testInviationShortLinkRestart :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testInviationShortLinkRestart ps = withAgentClients2 $ \a b -> do
   let userData = "some user data"
   (bId, CCLink connReq (Just shortLink)) <- withSmpServer ps $
@@ -1265,7 +1278,7 @@ testInviationShortLinkRestart ps = withAgentClients2 $ \a b -> do
     connReq' `shouldBe` connReq
     linkUserData connData' `shouldBe` userData
 
-testContactShortLinkRestart :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testContactShortLinkRestart :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testContactShortLinkRestart ps = withAgentClients2 $ \a b -> do
   let userData = "some user data"
   (contactId, CCLink connReq0 (Just shortLink)) <- withSmpServer ps $
@@ -1278,19 +1291,19 @@ testContactShortLinkRestart ps = withAgentClients2 $ \a b -> do
     connReq' `shouldBe` connReq
     linkUserData connData' `shouldBe` userData
     -- update user data
-    shortLink' <- runRight $ setContactShortLink a contactId updatedData
+    shortLink' <- runRight $ setContactShortLink a contactId updatedData Nothing
     shortLink' `shouldBe` shortLink
   withSmpServer ps $ do
     (connReq4, updatedConnData') <- runRight $ getConnShortLink b 1 shortLink
     connReq4 `shouldBe` connReq
     linkUserData updatedConnData' `shouldBe` updatedData
 
-testAddContactShortLinkRestart :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testAddContactShortLinkRestart :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testAddContactShortLinkRestart ps = withAgentClients2 $ \a b -> do
   let userData = "some user data"
   ((contactId, CCLink connReq0 Nothing), shortLink) <- withSmpServer ps $ runRight $ do
     r@(contactId, _) <- A.createConnection a 1 True SCMContact Nothing Nothing CR.IKPQOn SMOnlyCreate
-    (r,) <$> setContactShortLink a contactId userData
+    (r,) <$> setContactShortLink a contactId userData Nothing
   Right connReq <- pure $ smpDecode (smpEncode connReq0)
   let updatedData = "updated user data"
   withSmpServer ps $ do
@@ -1299,14 +1312,57 @@ testAddContactShortLinkRestart ps = withAgentClients2 $ \a b -> do
     connReq' `shouldBe` connReq
     linkUserData connData' `shouldBe` userData
     -- update user data
-    shortLink' <- runRight $ setContactShortLink a contactId updatedData
+    shortLink' <- runRight $ setContactShortLink a contactId updatedData Nothing
     shortLink' `shouldBe` shortLink
   withSmpServer ps $ do
     (connReq4, updatedConnData') <- runRight $ getConnShortLink b 1 shortLink
     connReq4 `shouldBe` connReq
     linkUserData updatedConnData' `shouldBe` updatedData
 
-testIncreaseConnAgentVersion :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testOldContactQueueShortLink :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
+testOldContactQueueShortLink ps@(_, msType) = withAgentClients2 $ \a b -> do
+  (contactId, CCLink connReq Nothing) <- withSmpServer ps $ runRight $
+    A.createConnection a 1 True SCMContact Nothing Nothing CR.IKPQOn SMOnlyCreate
+  -- make it an "old" queue
+  let updateStoreLog f = replaceSubstringInFile f " queue_mode=C" ""
+  () <- case testServerStoreConfig msType of
+    ASSCfg _ _ (SSCMemory (Just StorePaths {storeLogFile})) -> updateStoreLog storeLogFile
+    ASSCfg _ _ (SSCMemoryJournal {storeLogFile}) -> updateStoreLog storeLogFile
+    ASSCfg _ _ (SSCDatabaseJournal {storeCfg}) -> do
+#if defined(dbServerPostgres)
+      let AgentClient {agentEnv = Env {store}} = a
+      Right (SomeConn _ (ContactConnection _ RcvQueue {rcvId})) <- withTransaction store (`getConn` contactId)
+      st :: PostgresQueueStore (JournalQueue 'QSPostgres) <- newQueueStore @(JournalQueue 'QSPostgres) storeCfg
+      Right 1 <- runExceptT $ withDB' "test" st $ \db -> PSQL.execute db "UPDATE msg_queues SET queue_mode = ? WHERE recipient_id = ?" (Nothing :: Maybe QueueMode, rcvId)
+      closeQueueStore @(JournalQueue 'QSPostgres) st
+#else
+      error "no dbServerPostgres flag"
+#endif
+    _ -> pure ()
+
+  withSmpServer ps $ do
+    let userData = "some user data"
+    shortLink <- runRight $ setContactShortLink a contactId userData Nothing
+    (connReq', connData') <- runRight $ getConnShortLink b 1 shortLink
+    strDecode (strEncode shortLink) `shouldBe` Right shortLink
+    connReq' `shouldBe` connReq
+    linkUserData connData' `shouldBe` userData
+    -- update user data
+    let updatedData = "updated user data"
+    shortLink' <- runRight $ setContactShortLink a contactId updatedData Nothing
+    shortLink' `shouldBe` shortLink
+    -- check updated
+    (connReq'', updatedConnData') <- runRight $ getConnShortLink b 1 shortLink
+    connReq'' `shouldBe` connReq
+    linkUserData updatedConnData' `shouldBe` updatedData
+
+replaceSubstringInFile :: FilePath -> T.Text -> T.Text -> IO ()
+replaceSubstringInFile filePath oldText newText = do
+  content <- T.readFile filePath
+  let newContent = T.replace oldText newText content
+  T.writeFile filePath newContent
+
+testIncreaseConnAgentVersion :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testIncreaseConnAgentVersion ps = do
   alice <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
   bob <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB2
@@ -1371,7 +1427,7 @@ checkVersion c connId v = do
   ConnectionStats {connAgentVersion} <- getConnectionServers c connId
   liftIO $ connAgentVersion `shouldBe` VersionSMPA v
 
-testIncreaseConnAgentVersionMaxCompatible :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testIncreaseConnAgentVersionMaxCompatible :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testIncreaseConnAgentVersionMaxCompatible ps = do
   alice <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
   bob <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB2
@@ -1401,7 +1457,7 @@ testIncreaseConnAgentVersionMaxCompatible ps = do
     disposeAgentClient alice2
     disposeAgentClient bob2
 
-testIncreaseConnAgentVersionStartDifferentVersion :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testIncreaseConnAgentVersionStartDifferentVersion :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testIncreaseConnAgentVersionStartDifferentVersion ps = do
   alice <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
   bob <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
@@ -1427,7 +1483,7 @@ testIncreaseConnAgentVersionStartDifferentVersion ps = do
     disposeAgentClient alice2
     disposeAgentClient bob
 
-testDeliverClientRestart :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testDeliverClientRestart :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testDeliverClientRestart ps = do
   alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
@@ -1458,7 +1514,7 @@ testDeliverClientRestart ps = do
   disposeAgentClient alice
   disposeAgentClient bob2
 
-testDuplicateMessage :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testDuplicateMessage :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testDuplicateMessage ps = do
   alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
@@ -1510,7 +1566,7 @@ testDuplicateMessage ps = do
   disposeAgentClient alice2
   disposeAgentClient bob2
 
-testSkippedMessages :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testSkippedMessages :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testSkippedMessages (t, msType) = do
   alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
@@ -1561,7 +1617,7 @@ testSkippedMessages (t, msType) = do
   where
     cfg' = (cfgMS msType) {serverStoreCfg = ASSCfg SQSMemory SMSMemory $ SSCMemory $ Just $ StorePaths testStoreLogFile Nothing}
 
-testDeliveryAfterSubscriptionError :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testDeliveryAfterSubscriptionError :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testDeliveryAfterSubscriptionError ps = do
   (aId, bId) <- withAgentClients2 $ \a b -> do
     (aId, bId) <- withSmpServerStoreLogOn ps testPort $ \_ -> runRight $ makeConnection a b
@@ -1579,7 +1635,7 @@ testDeliveryAfterSubscriptionError ps = do
       withUP b aId $ \case ("", c, Msg "hello") -> c == aId; _ -> False
       ackMessage b aId 2 Nothing
 
-testMsgDeliveryQuotaExceeded :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testMsgDeliveryQuotaExceeded :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testMsgDeliveryQuotaExceeded ps =
   withAgentClients2 $ \a b -> withSmpServerStoreLogOn ps testPort $ \_ -> runRight_ $ do
     (aId, bId) <- makeConnection a b
@@ -1607,7 +1663,7 @@ testMsgDeliveryQuotaExceeded ps =
     get a =##> \case ("", c, SENT 6) -> bId == c; _ -> False
     liftIO $ concurrently_ (noMessages a "no more events") (noMessages b "no more events")
 
-testExpireMessage :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testExpireMessage :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testExpireMessage ps =
   withAgent 1 agentCfg {messageTimeout = 1.5, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB $ \a ->
     withAgent 2 agentCfg initAgentServers testDB2 $ \b -> do
@@ -1623,7 +1679,7 @@ testExpireMessage ps =
         withUP b aId $ \case ("", _, MsgErr 2 (MsgSkipped 2 2) "2") -> True; _ -> False
         ackMessage b aId 2 Nothing
 
-testExpireManyMessages :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testExpireManyMessages :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testExpireManyMessages ps =
   withAgent 1 agentCfg {messageTimeout = 2, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB $ \a ->
     withAgent 2 agentCfg initAgentServers testDB2 $ \b -> do
@@ -1662,7 +1718,7 @@ withUP a bId p =
         \case (corrId, c, AEvt SAEConn cmd) -> c == bId && p (corrId, c, cmd); _ -> False
       ]
 
-testExpireMessageQuota :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testExpireMessageQuota :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testExpireMessageQuota (t, msType) = withSmpServerConfigOn t (cfgMS msType) {msgQueueQuota = 1, maxJournalMsgCount = 2} testPort $ \_ -> do
   a <- getSMPAgentClient' 1 agentCfg {quotaExceededTimeout = 1, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB
   b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
@@ -1688,7 +1744,7 @@ testExpireMessageQuota (t, msType) = withSmpServerConfigOn t (cfgMS msType) {msg
     ackMessage b' aId 4 Nothing
   disposeAgentClient a
 
-testExpireManyMessagesQuota :: (ATransport, AStoreType) -> IO ()
+testExpireManyMessagesQuota :: (ASrvTransport, AStoreType) -> IO ()
 testExpireManyMessagesQuota (t, msType) = withSmpServerConfigOn t (cfgMS msType) {msgQueueQuota = 1, maxJournalMsgCount = 2} testPort $ \_ -> do
   a <- getSMPAgentClient' 1 agentCfg {quotaExceededTimeout = 2, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB
   b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
@@ -1725,7 +1781,7 @@ testExpireManyMessagesQuota (t, msType) = withSmpServerConfigOn t (cfgMS msType)
     ackMessage b' aId 4 Nothing
   disposeAgentClient a
 
-testRatchetSync :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testRatchetSync :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testRatchetSync ps = withAgentClients2 $ \alice bob ->
   withSmpServerStoreMsgLogOn ps testPort $ \_ -> do
     (aliceId, bobId, bob2) <- setupDesynchronizedRatchet alice bob
@@ -1799,7 +1855,7 @@ ratchetSyncP' cId rss = \case
     cId' == cId && rss' == rss && ratchetSyncState == rss
   _ -> False
 
-testRatchetSyncServerOffline :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testRatchetSyncServerOffline :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testRatchetSyncServerOffline ps = withAgentClients2 $ \alice bob -> do
   (aliceId, bobId, bob2) <- withSmpServerStoreMsgLogOn ps testPort $ \_ ->
     setupDesynchronizedRatchet alice bob
@@ -1825,7 +1881,7 @@ serverUpP = \case
   ("", "", AEvt SAENone (UP _ _)) -> True
   _ -> False
 
-testRatchetSyncClientRestart :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testRatchetSyncClientRestart :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testRatchetSyncClientRestart ps = do
   alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
@@ -1850,7 +1906,7 @@ testRatchetSyncClientRestart ps = do
   disposeAgentClient bob
   disposeAgentClient bob3
 
-testRatchetSyncSuspendForeground :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testRatchetSyncSuspendForeground :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testRatchetSyncSuspendForeground ps = do
   alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
@@ -1879,7 +1935,7 @@ testRatchetSyncSuspendForeground ps = do
   disposeAgentClient bob
   disposeAgentClient bob2
 
-testRatchetSyncSimultaneous :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testRatchetSyncSimultaneous :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testRatchetSyncSimultaneous ps = do
   alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
@@ -2006,7 +2062,7 @@ makeConnectionForUsers_ pqSupport sqSecured alice aliceUserId bob bobUserId = do
   get bob ##> ("", aliceId, A.CON pqEnc)
   pure (aliceId, bobId)
 
-testInactiveNoSubs :: (ATransport, AStoreType) -> IO ()
+testInactiveNoSubs :: (ASrvTransport, AStoreType) -> IO ()
 testInactiveNoSubs (t, msType) = do
   let cfg' = (cfgMS msType) {inactiveClientExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
   withSmpServerConfigOn t cfg' testPort $ \_ ->
@@ -2016,7 +2072,7 @@ testInactiveNoSubs (t, msType) = do
       Just (_, _, AEvt SAENone (DISCONNECT _ _)) <- timeout 5000000 $ atomically (readTBQueue $ subQ alice)
       pure ()
 
-testInactiveWithSubs :: (ATransport, AStoreType) -> IO ()
+testInactiveWithSubs :: (ASrvTransport, AStoreType) -> IO ()
 testInactiveWithSubs (t, msType) = do
   let cfg' = (cfgMS msType) {inactiveClientExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
   withSmpServerConfigOn t cfg' testPort $ \_ ->
@@ -2027,7 +2083,7 @@ testInactiveWithSubs (t, msType) = do
       -- and after 2 sec of inactivity no DOWN is sent as we have a live subscription
       liftIO $ timeout 1200000 (get alice) `shouldReturn` Nothing
 
-testActiveClientNotDisconnected :: (ATransport, AStoreType) -> IO ()
+testActiveClientNotDisconnected :: (ASrvTransport, AStoreType) -> IO ()
 testActiveClientNotDisconnected (t, msType) = do
   let cfg' = (cfgMS msType) {inactiveClientExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
   withSmpServerConfigOn t cfg' testPort $ \_ ->
@@ -2070,7 +2126,7 @@ testSuspendingAgent =
     liftIO $ foregroundAgent b
     get b =##> \case ("", c, Msg "hello 2") -> c == aId; _ -> False
 
-testSuspendingAgentCompleteSending :: (ATransport, AStoreType) -> IO ()
+testSuspendingAgentCompleteSending :: (ASrvTransport, AStoreType) -> IO ()
 testSuspendingAgentCompleteSending ps = withAgentClients2 $ \a b -> do
   (aId, bId) <- withSmpServerStoreLogOn ps testPort $ \_ -> runRight $ do
     (aId, bId) <- makeConnection a b
@@ -2101,7 +2157,7 @@ testSuspendingAgentCompleteSending ps = withAgentClients2 $ \a b -> do
     get a =##> \case ("", c, Msg "how are you?") -> c == bId; _ -> False
     ackMessage a bId 4 Nothing
 
-testSuspendingAgentTimeout :: (ATransport, AStoreType) -> IO ()
+testSuspendingAgentTimeout :: (ASrvTransport, AStoreType) -> IO ()
 testSuspendingAgentTimeout ps = withAgentClients2 $ \a b -> do
   (aId, _) <- withSmpServer ps . runRight $ do
     (aId, bId) <- makeConnection a b
@@ -2120,7 +2176,7 @@ testSuspendingAgentTimeout ps = withAgentClients2 $ \a b -> do
     ("", "", SUSPENDED) <- nGet b
     pure ()
 
-testBatchedSubscriptions :: Int -> Int -> (ATransport, AStoreType) -> IO ()
+testBatchedSubscriptions :: Int -> Int -> (ASrvTransport, AStoreType) -> IO ()
 testBatchedSubscriptions nCreate nDel ps@(t, ASType qsType _) =
   withAgentClientsCfgServers2 agentCfg agentCfg initAgentServers2 $ \a b -> do
     conns <- runServers $ do
@@ -2312,7 +2368,7 @@ testAsyncCommands sqSecured alice bob baseId =
   where
     msgId = subtract baseId
 
-testAsyncCommandsRestore :: (ATransport, AStoreType) -> IO ()
+testAsyncCommandsRestore :: (ASrvTransport, AStoreType) -> IO ()
 testAsyncCommandsRestore ps = do
   alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   bobId <- runRight $ createConnectionAsync alice 1 "1" True SCMInvitation (IKNoPQ PQSupportOn) SMSubscribe
@@ -2363,7 +2419,7 @@ testAcceptContactAsync sqSecured alice bob baseId =
   where
     msgId = subtract baseId
 
-testDeleteConnectionAsync :: (ATransport, AStoreType) -> IO ()
+testDeleteConnectionAsync :: (ASrvTransport, AStoreType) -> IO ()
 testDeleteConnectionAsync ps =
   withAgent 1 agentCfg {initialCleanupDelay = 10000, cleanupInterval = 10000, deleteErrorCount = 3} initAgentServers testDB $ \a -> do
     connIds <- withSmpServerStoreLogOn ps testPort $ \_ -> runRight $ do
@@ -2379,7 +2435,7 @@ testDeleteConnectionAsync ps =
       get a =##> \case ("", "", DEL_CONNS cs) -> length cs == 3 && all (`elem` connIds) cs; _ -> False
       liftIO $ noMessages a "nothing else should be delivered to alice"
 
-testWaitDeliveryNoPending :: (ATransport, AStoreType) -> IO ()
+testWaitDeliveryNoPending :: (ASrvTransport, AStoreType) -> IO ()
 testWaitDeliveryNoPending ps = withAgentClients2 $ \alice bob ->
   withSmpServerStoreLogOn ps testPort $ \_ -> runRight_ $ do
     (aliceId, bobId) <- makeConnection alice bob
@@ -2407,7 +2463,7 @@ testWaitDeliveryNoPending ps = withAgentClients2 $ \alice bob ->
     baseId = 1
     msgId = subtract baseId
 
-testWaitDelivery :: (ATransport, AStoreType) -> IO ()
+testWaitDelivery :: (ASrvTransport, AStoreType) -> IO ()
 testWaitDelivery ps =
   withAgent 1 agentCfg {initialCleanupDelay = 10000, cleanupInterval = 10000, deleteErrorCount = 3} initAgentServers testDB $ \alice ->
     withAgent 2 agentCfg initAgentServers testDB2 $ \bob -> do
@@ -2461,7 +2517,7 @@ testWaitDelivery ps =
     baseId = 1
     msgId = subtract baseId
 
-testWaitDeliveryAUTHErr :: (ATransport, AStoreType) -> IO ()
+testWaitDeliveryAUTHErr :: (ASrvTransport, AStoreType) -> IO ()
 testWaitDeliveryAUTHErr ps =
   withAgent 1 agentCfg {initialCleanupDelay = 10000, cleanupInterval = 10000, deleteErrorCount = 3} initAgentServers testDB $ \alice ->
     withAgent 2 agentCfg initAgentServers testDB2 $ \bob -> do
@@ -2504,7 +2560,7 @@ testWaitDeliveryAUTHErr ps =
     baseId = 1
     msgId = subtract baseId
 
-testWaitDeliveryTimeout :: (ATransport, AStoreType) -> IO ()
+testWaitDeliveryTimeout :: (ASrvTransport, AStoreType) -> IO ()
 testWaitDeliveryTimeout ps =
   withAgent 1 agentCfg {connDeleteDeliveryTimeout = 1, initialCleanupDelay = 10000, cleanupInterval = 10000, deleteErrorCount = 3} initAgentServers testDB $ \alice ->
     withAgent 2 agentCfg initAgentServers testDB2 $ \bob -> do
@@ -2544,7 +2600,7 @@ testWaitDeliveryTimeout ps =
     baseId = 1
     msgId = subtract baseId
 
-testWaitDeliveryTimeout2 :: (ATransport, AStoreType) -> IO ()
+testWaitDeliveryTimeout2 :: (ASrvTransport, AStoreType) -> IO ()
 testWaitDeliveryTimeout2 ps =
   withAgent 1 agentCfg {connDeleteDeliveryTimeout = 2, messageRetryInterval = fastMessageRetryInterval, initialCleanupDelay = 10000, cleanupInterval = 10000, deleteErrorCount = 3} initAgentServers testDB $ \alice ->
     withAgent 2 agentCfg initAgentServers testDB2 $ \bob -> do
@@ -2590,7 +2646,7 @@ testWaitDeliveryTimeout2 ps =
     baseId = 1
     msgId = subtract baseId
 
-testJoinConnectionAsyncReplyErrorV8 :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testJoinConnectionAsyncReplyErrorV8 :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testJoinConnectionAsyncReplyErrorV8 ps@(t, ASType qsType _) = do
   let initAgentServersSrv2 = initAgentServers {smp = userServers [testSMPServer2]}
   withAgent 1 cfg' initAgentServers testDB $ \a ->
@@ -2635,7 +2691,7 @@ testJoinConnectionAsyncReplyErrorV8 ps@(t, ASType qsType _) = do
           smpCfg = smpCfgVPrev {serverVRange = V.mkVersionRange minServerSMPRelayVersion sendingProxySMPVersion}  -- before SKEY
         }
 
-testJoinConnectionAsyncReplyError :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testJoinConnectionAsyncReplyError :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testJoinConnectionAsyncReplyError ps@(t, ASType qsType _) = do
   let initAgentServersSrv2 = initAgentServers {smp = userServers [testSMPServer2]}
   withAgent 1 agentCfg initAgentServers testDB $ \a ->
@@ -2702,7 +2758,7 @@ testDeleteUserQuietly =
     exchangeGreetingsMsgId 4 a bId b aId
     liftIO $ noMessages a "nothing else should be delivered to alice"
 
-testUsersNoServer :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testUsersNoServer :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testUsersNoServer ps = withAgentClientsCfg2 aCfg agentCfg $ \a b -> do
   (aId, bId, auId, _aId', bId') <- withSmpServerStoreLogOn ps testPort $ \_ -> runRight $ do
     (aId, bId) <- makeConnection a b
@@ -3137,7 +3193,7 @@ testCreateQueueAuth srvVersion clnt1 clnt2 sqSecured baseId = do
           sndAuthAlg = if srvVersion >= authCmdsSMPVersion && clntVersion >= authCmdsSMPVersion then C.AuthAlg C.SX25519 else C.AuthAlg C.SEd25519
        in getSMPAgentClient' clientId agentCfg {smpCfg, sndAuthAlg} servers db
 
-testSMPServerConnectionTest :: (ATransport, AStoreType) -> Maybe BasicAuth -> SMPServerWithAuth -> IO (Maybe ProtocolTestFailure)
+testSMPServerConnectionTest :: (ASrvTransport, AStoreType) -> Maybe BasicAuth -> SMPServerWithAuth -> IO (Maybe ProtocolTestFailure)
 testSMPServerConnectionTest (t, msType) newQueueBasicAuth srv =
   withSmpServerConfigOn t (cfgMS msType) {newQueueBasicAuth} testPort2 $ \_ -> do
     -- initially passed server is not running
@@ -3172,7 +3228,7 @@ testDeliveryReceipts =
     ackMessage b aId 5 (Just "") `catchError` \case (A.CMD PROHIBITED _) -> pure (); e -> liftIO $ expectationFailure ("unexpected error " <> show e)
     ackMessage b aId 5 Nothing
 
-testDeliveryReceiptsVersion :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testDeliveryReceiptsVersion :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testDeliveryReceiptsVersion ps = do
   a <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
   b <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
@@ -3225,7 +3281,7 @@ testDeliveryReceiptsVersion ps = do
     disposeAgentClient a'
     disposeAgentClient b'
 
-testDeliveryReceiptsConcurrent :: HasCallStack => (ATransport, AStoreType) -> IO ()
+testDeliveryReceiptsConcurrent :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testDeliveryReceiptsConcurrent (t, msType) =
   withSmpServerConfigOn t (cfgMS msType) {msgQueueQuota = 256, maxJournalMsgCount = 512} testPort $ \_ -> do
     withAgentClients2 $ \a b -> do
@@ -3577,7 +3633,7 @@ exchangeGreetingsMsgId_ :: HasCallStack => PQEncryption -> Int64 -> AgentClient
 exchangeGreetingsMsgId_ = exchangeGreetingsViaProxyMsgId_ False
 
 exchangeGreetingsViaProxy :: HasCallStack => Bool -> AgentClient -> ConnId -> AgentClient -> ConnId -> ExceptT AgentErrorType IO ()
-exchangeGreetingsViaProxy viaProxy = exchangeGreetingsViaProxyMsgId_ viaProxy PQEncOn 2 
+exchangeGreetingsViaProxy viaProxy = exchangeGreetingsViaProxyMsgId_ viaProxy PQEncOn 2
 
 exchangeGreetingsViaProxyMsgId_ :: HasCallStack => Bool -> PQEncryption -> Int64 -> AgentClient -> ConnId -> AgentClient -> ConnId -> ExceptT AgentErrorType IO ()
 exchangeGreetingsViaProxyMsgId_ viaProxy pqEnc msgId alice bobId bob aliceId = do

tests/AgentTests/MigrationTests.hs

@@ -11,7 +11,8 @@ import Simplex.Messaging.Agent.Store.Interface
 import Simplex.Messaging.Agent.Store.Migrations (migrationsToRun)
 import Simplex.Messaging.Agent.Store.Shared
 import System.Random (randomIO)
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
+import Util
 #if defined(dbPostgres)
 import qualified Data.ByteString.Char8 as B
 import Database.PostgreSQL.Simple (fromOnly)

tests/AgentTests/NotificationTests.hs

@@ -53,15 +53,12 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
-import Data.Text (Text)
-import qualified Data.Text as T
 import Data.Text.Encoding (encodeUtf8)
-import qualified Data.Text.IO as TIO
 import Data.Time.Clock.System (systemToUTCTime)
 import qualified Database.PostgreSQL.Simple as PSQL
 import NtfClient
 import SMPAgentClient (agentCfg, initAgentServers, initAgentServers2, testDB, testDB2, testNtfServer, testNtfServer2)
-import SMPClient (cfgMS, cfgJ2QS, cfgVPrev, ntfTestPort, ntfTestPort2, serverStoreConfig, testPort, testPort2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn, xit'')
+import SMPClient (cfgJ2QS, cfgMS, cfgVPrev, ntfTestPort, ntfTestPort2, testServerStoreConfig, testPort, testPort2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
 import Simplex.Messaging.Agent hiding (createConnection, joinConnection, sendMessage)
 import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..), withStore')
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig, Env (..), InitialAgentServers)
@@ -81,17 +78,18 @@ import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), NMsgMeta (..), NtfServer, ProtocolServer (..), SMPMsgMeta (..), SubscriptionMode (..))
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server.Env.STM (AStoreType (..), ServerConfig (..))
-import Simplex.Messaging.Transport (ATransport)
+import Simplex.Messaging.Transport (ASrvTransport)
 import System.Process (callCommand)
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
 import UnliftIO
+import Util
 #if defined(dbPostgres)
 import Database.PostgreSQL.Simple.SqlQQ (sql)
 #else
 import Database.SQLite.Simple.QQ (sql)
 #endif
 
-notificationTests :: (ATransport, AStoreType) -> Spec
+notificationTests :: (ASrvTransport, AStoreType) -> Spec
 notificationTests ps@(t, _) = do
   describe "Managing notification tokens" $ do
     it "should register and verify notification token" $
@@ -156,10 +154,10 @@ notificationTests ps@(t, _) = do
     it "should resume subscriptions after SMP server is restarted" $
       withAPNSMockServer $ \apns ->
         withNtfServer t $ testNotificationsSMPRestart ps apns
-  describe "Notifications after SMP server restart" $
+  describe "Notifications after SMP server restart (batched)" $
     it "should resume batched subscriptions after SMP server is restarted" $
       withAPNSMockServer $ \apns ->
-        withNtfServer t $ testNotificationsSMPRestartBatch 100 ps apns
+        withNtfServer t $ testNotificationsSMPRestartBatch 50 ps apns
   describe "should switch notifications to the new queue" $
     testServerMatrix2 ps $ \servers ->
       withAPNSMockServer $ \apns ->
@@ -175,7 +173,7 @@ notificationTests ps@(t, _) = do
         withNtfServerOn t ntfTestPort2 ntfTestDBCfg2 . withNtfServerThreadOn t ntfTestPort ntfTestDBCfg $ \ntf ->
           testNotificationsNewToken apns ntf
 
-testNtfMatrix :: HasCallStack => (ATransport, AStoreType) -> (APNSMockServer -> AgentMsgId -> AgentClient -> AgentClient -> IO ()) -> Spec
+testNtfMatrix :: HasCallStack => (ASrvTransport, AStoreType) -> (APNSMockServer -> AgentMsgId -> AgentClient -> AgentClient -> IO ()) -> Spec
 testNtfMatrix ps@(_, msType) runTest = do
   describe "next and current" $ do
     it "curr servers; curr clients" $ runNtfTestCfg ps 1 cfg' ntfServerCfg agentCfg agentCfg runTest
@@ -192,9 +190,9 @@ testNtfMatrix ps@(_, msType) runTest = do
     cfg' = cfgMS msType
     cfgVPrev' = cfgVPrev msType
 
-runNtfTestCfg :: HasCallStack => (ATransport, AStoreType) -> AgentMsgId -> ServerConfig -> NtfServerConfig -> AgentConfig -> AgentConfig -> (APNSMockServer -> AgentMsgId -> AgentClient -> AgentClient -> IO ()) -> IO ()
+runNtfTestCfg :: HasCallStack => (ASrvTransport, AStoreType) -> AgentMsgId -> ServerConfig -> NtfServerConfig -> AgentConfig -> AgentConfig -> (APNSMockServer -> AgentMsgId -> AgentClient -> AgentClient -> IO ()) -> IO ()
 runNtfTestCfg (t, msType) baseId smpCfg ntfCfg aCfg bCfg runTest = do
-  let smpCfg' = smpCfg {serverStoreCfg = serverStoreConfig msType}
+  let smpCfg' = smpCfg {serverStoreCfg = testServerStoreConfig msType}
   withSmpServerConfigOn t smpCfg' testPort $ \_ ->
     withAPNSMockServer $ \apns ->
       withNtfServerCfg ntfCfg {transports = [(ntfTestPort, t, False)]} $ \_ ->
@@ -227,8 +225,6 @@ v .-> key = do
 
 testNtfTokenRepeatRegistration :: APNSMockServer -> IO ()
 testNtfTokenRepeatRegistration apns = do
-  -- setLogLevel LogError -- LogDebug
-  -- withGlobalLogging logCfg $ do
   withAgent 1 agentCfg initAgentServers testDB $ \a -> runRight_ $ do
     let tkn = DeviceToken PPApnsTest "abcd"
     NTRegistered <- registerNtfToken a tkn NMPeriodic
@@ -248,8 +244,6 @@ testNtfTokenRepeatRegistration apns = do
 
 testNtfTokenSecondRegistration :: APNSMockServer -> IO ()
 testNtfTokenSecondRegistration apns =
-  -- setLogLevel LogError -- LogDebug
-  -- withGlobalLogging logCfg $ do
   withAgentClients2 $ \a a' -> runRight_ $ do
     let tkn = DeviceToken PPApnsTest "abcd"
     NTRegistered <- registerNtfToken a tkn NMPeriodic
@@ -278,7 +272,7 @@ testNtfTokenSecondRegistration apns =
     NTActive <- checkNtfToken a' tkn
     pure ()
 
-testNtfTokenServerRestart :: ATransport -> APNSMockServer -> IO ()
+testNtfTokenServerRestart :: ASrvTransport -> APNSMockServer -> IO ()
 testNtfTokenServerRestart t apns = do
   let tkn = DeviceToken PPApnsTest "abcd"
   ntfData <- withAgent 1 agentCfg initAgentServers testDB $ \a ->
@@ -299,7 +293,7 @@ testNtfTokenServerRestart t apns = do
       NTActive <- checkNtfToken a' tkn
       pure ()
 
-testNtfTokenServerRestartReverify :: ATransport -> APNSMockServer -> IO ()
+testNtfTokenServerRestartReverify :: ASrvTransport -> APNSMockServer -> IO ()
 testNtfTokenServerRestartReverify t apns = do
   let tkn = DeviceToken PPApnsTest "abcd"
   withAgent 1 agentCfg initAgentServers testDB $ \a -> do
@@ -322,7 +316,7 @@ testNtfTokenServerRestartReverify t apns = do
       NTActive <- checkNtfToken a' tkn
       pure ()
 
-testNtfTokenServerRestartReverifyTimeout :: ATransport -> APNSMockServer -> IO ()
+testNtfTokenServerRestartReverifyTimeout :: ASrvTransport -> APNSMockServer -> IO ()
 testNtfTokenServerRestartReverifyTimeout t apns = do
   let tkn = DeviceToken PPApnsTest "abcd"
   withAgent 1 agentCfg initAgentServers testDB $ \a@AgentClient {agentEnv = Env {store}} -> do
@@ -357,7 +351,7 @@ testNtfTokenServerRestartReverifyTimeout t apns = do
       NTActive <- checkNtfToken a' tkn
       pure ()
 
-testNtfTokenServerRestartReregister :: ATransport -> APNSMockServer -> IO ()
+testNtfTokenServerRestartReregister :: ASrvTransport -> APNSMockServer -> IO ()
 testNtfTokenServerRestartReregister t apns = do
   let tkn = DeviceToken PPApnsTest "abcd"
   withAgent 1 agentCfg initAgentServers testDB $ \a ->
@@ -381,7 +375,7 @@ testNtfTokenServerRestartReregister t apns = do
       NTActive <- checkNtfToken a' tkn
       pure ()
 
-testNtfTokenServerRestartReregisterTimeout :: ATransport -> APNSMockServer -> IO ()
+testNtfTokenServerRestartReregisterTimeout :: ASrvTransport -> APNSMockServer -> IO ()
 testNtfTokenServerRestartReregisterTimeout t apns = do
   let tkn = DeviceToken PPApnsTest "abcd"
   withAgent 1 agentCfg initAgentServers testDB $ \a@AgentClient {agentEnv = Env {store}} -> do
@@ -422,7 +416,7 @@ getTestNtfTokenPort a =
     Just NtfToken {ntfServer = ProtocolServer {port}} -> pure port
     Nothing -> error "no active NtfToken"
 
-testNtfTokenMultipleServers :: ATransport -> APNSMockServer -> IO ()
+testNtfTokenMultipleServers :: ASrvTransport -> APNSMockServer -> IO ()
 testNtfTokenMultipleServers t apns = do
   let tkn = DeviceToken PPApnsTest "abcd"
   withAgent 1 agentCfg initAgentServers2 testDB $ \a ->
@@ -446,7 +440,7 @@ testNtfTokenMultipleServers t apns = do
         Left _ <- tryError (checkNtfToken a tkn)
         pure ()
 
-testNtfTokenChangeServers :: ATransport -> APNSMockServer -> IO ()
+testNtfTokenChangeServers :: ASrvTransport -> APNSMockServer -> IO ()
 testNtfTokenChangeServers t apns =
   withNtfServerThreadOn t ntfTestPort ntfTestDBCfg $ \ntf -> do
     tkn1 <- withAgent 1 agentCfg initAgentServers testDB $ \a -> runRight $ do
@@ -476,7 +470,7 @@ testNtfTokenChangeServers t apns =
         tkn <- registerTestToken a "qwer" NMInstant apns
         checkNtfToken a tkn >>= \r -> liftIO $ r `shouldBe` NTActive
 
-testNtfTokenReRegisterInvalid :: ATransport -> APNSMockServer -> IO ()
+testNtfTokenReRegisterInvalid :: ASrvTransport -> APNSMockServer -> IO ()
 testNtfTokenReRegisterInvalid t apns = do
   tkn <- withNtfServer t $ do
     withAgent 1 agentCfg initAgentServers testDB $ \a -> runRight $ do
@@ -501,13 +495,7 @@ testNtfTokenReRegisterInvalid t apns = do
       NTActive <- checkNtfToken a tkn1
       pure ()
 
-replaceSubstringInFile :: FilePath -> Text -> Text -> IO ()
-replaceSubstringInFile filePath oldText newText = do
-  content <- TIO.readFile filePath
-  let newContent = T.replace oldText newText content
-  TIO.writeFile filePath newContent
-
-testNtfTokenReRegisterInvalidOnCheck :: ATransport -> APNSMockServer -> IO ()
+testNtfTokenReRegisterInvalidOnCheck :: ASrvTransport -> APNSMockServer -> IO ()
 testNtfTokenReRegisterInvalidOnCheck t apns = do
   tkn <- withNtfServer t $ do
     withAgent 1 agentCfg initAgentServers testDB $ \a -> runRight $ do
@@ -532,7 +520,7 @@ testNtfTokenReRegisterInvalidOnCheck t apns = do
       NTActive <- checkNtfToken a tkn1
       pure ()
 
-testRunNTFServerTests :: ATransport -> NtfServer -> IO (Maybe ProtocolTestFailure)
+testRunNTFServerTests :: ASrvTransport -> NtfServer -> IO (Maybe ProtocolTestFailure)
 testRunNTFServerTests t srv =
   withNtfServer t $
     withAgent 1 agentCfg initAgentServers testDB $ \a ->
@@ -559,7 +547,6 @@ testNotificationSubscriptionExistingConnection apns baseId alice@AgentClient {ag
     verifyNtfToken alice tkn vNonce verification
     NTActive <- checkNtfToken alice tkn
     -- send message
-    liftIO $ threadDelay 250000
     1 <- msgId <$> sendMessage bob aliceId (SMP.MsgFlags True) "hello"
     get bob ##> ("", aliceId, SENT $ baseId + 1)
     -- notification
@@ -571,11 +558,10 @@ testNotificationSubscriptionExistingConnection apns baseId alice@AgentClient {ag
   -- alice client already has subscription for the connection,
   [Left (CMD PROHIBITED _)] <- getConnectionMessages alice [ConnMsgReq cId 1 $ Just $ systemToUTCTime msgTs]
 
-  threadDelay 500000
+  threadDelay 1000000
   suspendAgent alice 0
   closeDBStore store
   threadDelay 1000000 >> callCommand "sync" >> threadDelay 1000000
-  putStrLn "before opening the database from another agent"
 
   -- aliceNtf client doesn't have subscription and is allowed to get notification message
   withAgent 3 aliceCfg initAgentServers testDB $ \aliceNtf -> do
@@ -583,7 +569,6 @@ testNotificationSubscriptionExistingConnection apns baseId alice@AgentClient {ag
     pure ()
 
   threadDelay 1000000 >> callCommand "sync" >> threadDelay 1000000
-  putStrLn "after closing the database in another agent"
   reopenDBStore store
   foregroundAgent alice
   threadDelay 500000
@@ -757,7 +742,7 @@ testChangeToken apns = withAgent 1 agentCfg initAgentServers testDB2 $ \bob -> d
     baseId = 1
     msgId = subtract baseId
 
-testNotificationsStoreLog :: (ATransport, AStoreType) -> APNSMockServer -> IO ()
+testNotificationsStoreLog :: (ASrvTransport, AStoreType) -> APNSMockServer -> IO ()
 testNotificationsStoreLog ps@(t, _) apns = withAgentClients2 $ \alice bob -> do
   withSmpServerStoreMsgLogOn ps testPort $ \_ -> do
     (aliceId, bobId) <- withNtfServer t $ runRight $ do
@@ -792,7 +777,7 @@ testNotificationsStoreLog ps@(t, _) apns = withAgentClients2 $ \alice bob -> do
     withNtfServer t $ runRight_ $ do
       void $ messageNotificationData alice apns
 
-testNotificationsSMPRestart :: (ATransport, AStoreType) -> APNSMockServer -> IO ()
+testNotificationsSMPRestart :: (ASrvTransport, AStoreType) -> APNSMockServer -> IO ()
 testNotificationsSMPRestart ps apns = withAgentClients2 $ \alice bob -> do
   (aliceId, bobId) <- withSmpServerStoreLogOn ps testPort $ \threadId -> runRight $ do
     (aliceId, bobId) <- makeConnection alice bob
@@ -820,7 +805,7 @@ testNotificationsSMPRestart ps apns = withAgentClients2 $ \alice bob -> do
     get alice =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False
     liftIO $ killThread threadId
 
-testNotificationsSMPRestartBatch :: Int -> (ATransport, AStoreType) -> APNSMockServer -> IO ()
+testNotificationsSMPRestartBatch :: Int -> (ASrvTransport, AStoreType) -> APNSMockServer -> IO ()
 testNotificationsSMPRestartBatch n ps@(t, ASType qsType _) apns =
   withAgentClientsCfgServers2 agentCfg agentCfg initAgentServers2 $ \a b -> do
     threadDelay 1000000

tests/AgentTests/SQLiteTests.hs

@@ -52,11 +52,13 @@ import Simplex.Messaging.Crypto.File (CryptoFile (..))
 import Simplex.Messaging.Crypto.Ratchet (InitialKeys (..), pattern PQSupportOn)
 import qualified Simplex.Messaging.Crypto.Ratchet as CR
 import Simplex.Messaging.Encoding.String (StrEncoding (..))
-import Simplex.Messaging.Protocol (EntityId (..), SubscriptionMode (..), QueueMode (..), pattern VersionSMPC)
+import Simplex.Messaging.Protocol (EntityId (..), QueueMode (..), SubscriptionMode (..), pattern VersionSMPC)
 import qualified Simplex.Messaging.Protocol as SMP
+import Simplex.Messaging.Agent.Store.Entity
 import System.Random
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
 import UnliftIO.Directory (removeFile)
+import Util
 
 testDB :: String
 testDB = "tests/tmp/smp-agent.test.db"
@@ -229,7 +231,7 @@ rcvQueue1 =
       queueMode = Just QMMessaging,
       shortLink = Nothing,
       status = New,
-      dbQueueId = DBNewQueue,
+      dbQueueId = DBNewEntity,
       primary = True,
       dbReplaceQueueId = Nothing,
       rcvSwchStatus = Nothing,
@@ -251,7 +253,7 @@ sndQueue1 =
       e2ePubKey = Nothing,
       e2eDhSecret = testDhSecret,
       status = New,
-      dbQueueId = DBNewQueue,
+      dbQueueId = DBNewEntity,
       primary = True,
       dbReplaceQueueId = Nothing,
       sndSwchStatus = Nothing,
@@ -270,11 +272,11 @@ testCreateRcvConn =
     g <- C.newRandom
     Right (connId, rq@RcvQueue {dbQueueId}) <- createRcvConn db g cData1 rcvQueue1 SCMInvitation
     connId `shouldBe` "conn1"
-    dbQueueId `shouldBe` DBQueueId 1
+    dbQueueId `shouldBe` DBEntityId 1
     getConn db "conn1"
       `shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 rq))
     Right sq@SndQueue {dbQueueId = dbQueueId'} <- upgradeRcvConnToDuplex db "conn1" sndQueue1
-    dbQueueId' `shouldBe` DBQueueId 1
+    dbQueueId' `shouldBe` DBEntityId 1
     getConn db "conn1"
       `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 [rq] [sq]))
 
@@ -286,7 +288,7 @@ testCreateRcvConnRandomId =
     getConn db connId
       `shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 {connId} rq))
     Right sq@SndQueue {dbQueueId = dbQueueId'} <- upgradeRcvConnToDuplex db connId sndQueue1
-    dbQueueId' `shouldBe` DBQueueId 1
+    dbQueueId' `shouldBe` DBEntityId 1
     getConn db connId
       `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 {connId} [rq] [sq]))
 
@@ -304,11 +306,11 @@ testCreateSndConn =
     g <- C.newRandom
     Right (connId, sq@SndQueue {dbQueueId}) <- createSndConn db g cData1 sndQueue1
     connId `shouldBe` "conn1"
-    dbQueueId `shouldBe` DBQueueId 1
+    dbQueueId `shouldBe` DBEntityId 1
     getConn db "conn1"
       `shouldReturn` Right (SomeConn SCSnd (SndConnection cData1 sq))
     Right rq@RcvQueue {dbQueueId = dbQueueId'} <- upgradeSndConnToDuplex db "conn1" rcvQueue1
-    dbQueueId' `shouldBe` DBQueueId 1
+    dbQueueId' `shouldBe` DBEntityId 1
     getConn db "conn1"
       `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 [rq] [sq]))
 
@@ -320,7 +322,7 @@ testCreateSndConnRandomID =
     getConn db connId
       `shouldReturn` Right (SomeConn SCSnd (SndConnection cData1 {connId} sq))
     Right (rq@RcvQueue {dbQueueId = dbQueueId'}) <- upgradeSndConnToDuplex db connId rcvQueue1
-    dbQueueId' `shouldBe` DBQueueId 1
+    dbQueueId' `shouldBe` DBEntityId 1
     getConn db connId
       `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 {connId} [rq] [sq]))
 
@@ -411,7 +413,7 @@ testUpgradeRcvConnToDuplex =
               e2ePubKey = Nothing,
               e2eDhSecret = testDhSecret,
               status = New,
-              dbQueueId = DBNewQueue,
+              dbQueueId = DBNewEntity,
               sndSwchStatus = Nothing,
               primary = True,
               dbReplaceQueueId = Nothing,
@@ -442,7 +444,7 @@ testUpgradeSndConnToDuplex =
               queueMode = Just QMMessaging,
               shortLink = Nothing,
               status = New,
-              dbQueueId = DBNewQueue,
+              dbQueueId = DBNewEntity,
               rcvSwchStatus = Nothing,
               primary = True,
               dbReplaceQueueId = Nothing,

tests/AgentTests/SchemaDump.hs

@@ -18,7 +18,8 @@ import Simplex.Messaging.Agent.Store.Shared (Migration (..), MigrationConfirmati
 import Simplex.Messaging.Util (ifM)
 import System.Directory (doesFileExist, removeFile)
 import System.Process (readCreateProcess, shell)
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
+import Util
 
 testDB :: FilePath
 testDB = "tests/tmp/test_agent_schema.db"

tests/AgentTests/ServerChoice.hs

@@ -14,8 +14,9 @@ import Simplex.Messaging.Agent.Client hiding (userServers)
 import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Client (defaultNetworkConfig)
 import Simplex.Messaging.Protocol
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
 import Test.QuickCheck
+import Util
 import XFTPClient (testXFTPServer)
 
 serverChoiceTests :: Spec

tests/AgentTests/ShortLinkTests.hs

@@ -11,7 +11,8 @@ import Control.Monad.Except
 import Simplex.Messaging.Agent.Protocol (AgentErrorType (..), ConnectionMode (..), LinkKey (..), SMPAgentError (..), linkUserData, supportedSMPAgentVRange)
 import qualified Simplex.Messaging.Crypto as C
 import qualified Simplex.Messaging.Crypto.ShortLink as SL
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
+import Util
 
 shortLinkTests :: Spec
 shortLinkTests = do
@@ -20,7 +21,7 @@ shortLinkTests = do
     it "should fail to decrypt invitation data with bad hash" testInvShortLinkBadDataHash
   describe "contact short link" $ do
     it "should encrypt and decrypt data" testContactShortLink
-    it "should encrypt updated user data" testUpdateContactShortLink    
+    it "should encrypt updated user data" testUpdateContactShortLink
     it "should fail to decrypt contact data with bad hash" testContactShortLinkBadDataHash
     it "should fail to decrypt contact data with bad signature" testContactShortLinkBadSignature
 

tests/CLITests.hs

@@ -1,4 +1,5 @@
 {-# LANGUAGE CPP #-}
+{-# LANGUAGE DataKinds #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE NamedFieldPuns #-}
@@ -23,7 +24,7 @@ import qualified Network.HTTP2.Client as H2
 import Simplex.FileTransfer.Server.Main (xftpServerCLI)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Server.Main (smpServerCLI, smpServerCLI_)
-import Simplex.Messaging.Transport (TLS (..), defaultSupportedParams, defaultSupportedParamsHTTPS, simplexMQVersion, supportedClientSMPRelayVRange)
+import Simplex.Messaging.Transport (TLS (..), TransportPeer (..), defaultSupportedParams, defaultSupportedParamsHTTPS, simplexMQVersion, supportedClientSMPRelayVRange)
 import Simplex.Messaging.Transport.Client (TransportClientConfig (..), defaultTransportClientConfig, runTLSTransportClient, smpClientHandshake)
 import Simplex.Messaging.Transport.HTTP2 (HTTP2Body (..))
 import qualified Simplex.Messaging.Transport.HTTP2.Client as HC
@@ -35,12 +36,13 @@ import System.Environment (withArgs)
 import System.FilePath ((</>))
 import System.IO.Silently (capture_)
 import System.Timeout (timeout)
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
 import Test.Main (withStdin)
 import UnliftIO (catchAny)
 import UnliftIO.Async (async, cancel)
 import UnliftIO.Concurrent (threadDelay)
 import UnliftIO.Exception (bracket)
+import Util
 
 #if defined(dbServerPostgres)
 import qualified Database.PostgreSQL.Simple as PSQL
@@ -191,9 +193,9 @@ smpServerTestStatic = do
       runRight_ . void $ smpClientHandshake tls Nothing caSMP supportedClientSMPRelayVRange False
     logDebug "Combined SMP works"
   where
-    getCerts :: TLS -> [X.Certificate]
+    getCerts :: TLS 'TClient -> [X.Certificate]
     getCerts tls =
-      let X.CertificateChain cc = tlsServerCerts tls
+      let X.CertificateChain cc = tlsPeerCert tls
        in map (X.signedObject . X.getSigned) cc
 
 #if defined(dbServerPostgres)

tests/CoreTests/BatchingTests.hs

@@ -24,7 +24,8 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Transport
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
+import Util
 
 batchingTests :: Spec
 batchingTests = do

tests/CoreTests/CryptoFileTests.hs

@@ -13,7 +13,8 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.File (CryptoFile (..), FTCryptoError (..))
 import qualified Simplex.Messaging.Crypto.File as CF
 import System.Directory (getFileSize)
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
+import Util
 
 cryptoFileTests :: Spec
 cryptoFileTests = do

tests/CoreTests/CryptoTests.hs

@@ -24,9 +24,10 @@ import qualified Simplex.Messaging.Crypto as C
 import qualified Simplex.Messaging.Crypto.Lazy as LC
 import Simplex.Messaging.Crypto.SNTRUP761.Bindings
 import Simplex.Messaging.Transport.Client
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
 import Test.Hspec.QuickCheck (modifyMaxSuccess)
 import Test.QuickCheck
+import Util
 
 cryptoTests :: Spec
 cryptoTests = do

tests/CoreTests/EncodingTests.hs

@@ -16,9 +16,10 @@ import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Transport.Client (TransportHost (..))
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
 import Test.Hspec.QuickCheck (modifyMaxSuccess)
 import Test.QuickCheck
+import Util
 
 int64 :: Int64
 int64 = 1234567890123456789

tests/CoreTests/MsgStoreTests.hs

@@ -10,8 +10,8 @@
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StandaloneDeriving #-}
 {-# LANGUAGE TypeApplications #-}
-{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
 {-# OPTIONS_GHC -Wno-orphans #-}
+{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
 
 module CoreTests.MsgStoreTests where
 
@@ -23,13 +23,14 @@ import Control.Monad
 import Control.Monad.IO.Class
 import Control.Monad.Trans.Except
 import Crypto.Random (ChaChaDRG)
+import qualified Data.ByteString.Base64.URL as B64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import qualified Data.ByteString.Base64.URL as B64
 import Data.List (isPrefixOf, isSuffixOf)
 import Data.Maybe (fromJust)
 import Data.Time.Clock (addUTCTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
+import SMPClient (testStoreLogFile, testStoreMsgsDir, testStoreMsgsDir2, testStoreMsgsFile, testStoreMsgsFile2)
 import Simplex.Messaging.Crypto (pattern MaxLenBS)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol (EntityId (..), LinkId, Message (..), QueueLinkData, RecipientId, SParty (..), noMsgFlags)
@@ -43,11 +44,11 @@ import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.QueueStore.QueueInfo
 import Simplex.Messaging.Server.QueueStore.Types
 import Simplex.Messaging.Server.StoreLog (closeStoreLog, logCreateQueue)
-import SMPClient (testStoreLogFile, testStoreMsgsDir, testStoreMsgsDir2, testStoreMsgsFile, testStoreMsgsFile2)
 import System.Directory (copyFile, createDirectoryIfMissing, listDirectory, removeFile, renameFile)
 import System.FilePath ((</>))
 import System.IO (IOMode (..), withFile)
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
+import Util
 
 msgStoreTests :: Spec
 msgStoreTests = do
@@ -256,7 +257,6 @@ testQueueState ms = do
   length . lines <$> readFile statePath `shouldReturn` 1
   readQueueState ms statePath `shouldReturn` (Just state, False)
   length <$> listDirectory dir `shouldReturn` 1 -- no backup
-
   let state1 =
         state
           { size = 1,
@@ -267,7 +267,6 @@ testQueueState ms = do
   length . lines <$> readFile statePath `shouldReturn` 2
   readQueueState ms statePath `shouldReturn` (Just state1, False)
   length <$> listDirectory dir `shouldReturn` 1 -- no backup
-
   let state2 =
         state
           { size = 2,
@@ -343,7 +342,7 @@ testRemoveJournals ms = do
   runRight $ do
     q <- ExceptT $ addQueue ms rId qr
     Just (Message {msgId = mId1}, True) <- write q "message 1"
-    Just (Message {msgId = mId2}, False) <- write q "message 2"    
+    Just (Message {msgId = mId2}, False) <- write q "message 2"
     (Msg "message 1", Msg "message 2") <- tryDelPeekMsg ms q mId1
     (Msg "message 2", Nothing) <- tryDelPeekMsg ms q mId2
     liftIO $ closeMsgQueue ms q

tests/CoreTests/RetryIntervalTests.hs

@@ -8,14 +8,15 @@ import Control.Concurrent.STM
 import Control.Monad (when)
 import Data.Time.Clock (UTCTime, diffUTCTime, getCurrentTime, nominalDiffTimeToSeconds)
 import Simplex.Messaging.Agent.RetryInterval
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
+import Util
 
 retryIntervalTests :: Spec
 retryIntervalTests = do
   describe "Retry interval with 2 modes and lock" $ do
     testRetryIntervalSameMode
     testRetryIntervalSwitchMode
-  describe "Foreground retry interval" $ do 
+  describe "Foreground retry interval" $ do
     testRetryForeground
     testRetryToBackground
     testRetrySkipWhenForeground
@@ -103,7 +104,7 @@ testRetryForeground =
       when (length ints < 8) $ loop
     (reverse <$> readTVarIO intervals) `shouldReturn` [0, 1, 1, 1, 2, 3, 4, 4]
     (reverse <$> readTVarIO reportedIntervals)
-      `shouldReturn` [ 10000, 10000, 15000, 22500, 33750, 40000, 40000, 40000]
+      `shouldReturn` [10000, 10000, 15000, 22500, 33750, 40000, 40000, 40000]
 
 testRetryToBackground :: Spec
 testRetryToBackground =
@@ -124,7 +125,7 @@ testRetryToBackground =
       )
     (reverse <$> readTVarIO intervals) `shouldReturn` [0, 1, 1, 1, 2, 3, 4, 4]
     (reverse <$> readTVarIO reportedIntervals)
-      `shouldReturn` [ 10000, 10000, 15000, 22500, 33750, 40000, 40000, 40000]
+      `shouldReturn` [10000, 10000, 15000, 22500, 33750, 40000, 40000, 40000]
 
 testRetrySkipWhenForeground :: Spec
 testRetrySkipWhenForeground =
@@ -149,7 +150,7 @@ testRetrySkipWhenForeground =
       )
     (reverse <$> readTVarIO intervals) `shouldReturn` [0, 1, 1, 1, 2, 0, 1, 1, 1, 2, 3, 1]
     (reverse <$> readTVarIO reportedIntervals)
-      `shouldReturn` [ 10000, 10000, 15000, 22500, 33750, 10000, 10000, 15000, 22500, 33750, 40000, 10000]
+      `shouldReturn` [10000, 10000, 15000, 22500, 33750, 10000, 10000, 15000, 22500, 33750, 40000, 10000]
 
 addInterval :: TVar [Int] -> TVar UTCTime -> IO [Int]
 addInterval intervals ts = do

tests/CoreTests/SOCKSSettings.hs

@@ -12,7 +12,8 @@ import Simplex.Messaging.Client
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (ErrorType)
 import Simplex.Messaging.Transport.Client
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
+import Util
 
 socksSettingsTests :: Spec
 socksSettingsTests = do

tests/CoreTests/StoreLogTests.hs

@@ -29,7 +29,8 @@ import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.QueueStore.STM (STMQueueStore (..))
 import Simplex.Messaging.Server.QueueStore.Types
 import Simplex.Messaging.Server.StoreLog
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
+import Util
 
 testPublicAuthKey :: C.APublicAuthKey
 testPublicAuthKey = C.APublicAuthKey C.SEd25519 (C.publicKey "MC4CAQAwBQYDK2VwBCIEIDfEfevydXXfKajz3sRkcQ7RPvfWUPoq6pu1TYHV1DEe")
@@ -81,19 +82,19 @@ storeLogTests =
             saved = [CreateQueue rId' qr'],
             compacted = [CreateQueue rId' qr'],
             state = M.fromList [(rId', qr')]
-          },          
+          },
         SLTC
           { name = "create new queue, add link data",
             saved = [CreateQueue rId' qr' {queueData = Nothing}, CreateLink rId' lnkId qd],
             compacted = [CreateQueue rId' qr'],
             state = M.fromList [(rId', qr')]
-          },          
+          },
         SLTC
           { name = "create new queue with link data, delete data",
             saved = [CreateQueue rId' qr', DeleteLink rId'],
             compacted = [CreateQueue rId' qr' {queueData = Nothing}],
             state = M.fromList [(rId', qr' {queueData = Nothing})]
-          },          
+          },
         SLTC
           { name = "secure queue",
             saved = [CreateQueue rId qr, SecureQueue rId testPublicAuthKey],

tests/CoreTests/TRcvQueuesTests.hs

@@ -14,12 +14,14 @@ import qualified Data.Map as M
 import qualified Data.Set as S
 import Data.String (IsString (..))
 import Simplex.Messaging.Agent.Protocol (ConnId, QueueStatus (..), UserId)
-import Simplex.Messaging.Agent.Store (DBQueueId (..), RcvQueue, StoredRcvQueue (..))
+import Simplex.Messaging.Agent.Store (RcvQueue, StoredRcvQueue (..))
+import Simplex.Messaging.Agent.Store.Entity
 import qualified Simplex.Messaging.Agent.TRcvQueues as RQ
 import qualified Simplex.Messaging.Crypto as C
-import Simplex.Messaging.Protocol (EntityId (..), RecipientId, SMPServer, QueueMode (..), pattern NoEntity, pattern VersionSMPC)
-import Test.Hspec
+import Simplex.Messaging.Protocol (EntityId (..), QueueMode (..), RecipientId, SMPServer, pattern NoEntity, pattern VersionSMPC)
+import Test.Hspec hiding (fit, it)
 import UnliftIO
+import Util
 
 tRcvQueuesTests :: Spec
 tRcvQueuesTests = do
@@ -120,7 +122,7 @@ getSessQueuesTest = do
   atomically (RQ.hasSessQueues tSess3 trq) `shouldReturn` False
   let tSess4 = (0, "smp://1234-w==@alpha", Nothing)
   RQ.getSessQueues tSess4 trq `shouldReturn` [dummyRQ 0 "smp://1234-w==@alpha" "c2" "r2", dummyRQ 0 "smp://1234-w==@alpha" "c1" "r1"]
-  atomically (RQ.hasSessQueues tSess4 trq) `shouldReturn`True
+  atomically (RQ.hasSessQueues tSess4 trq) `shouldReturn` True
 
 getDelSessQueuesTest :: IO ()
 getDelSessQueuesTest = do
@@ -200,7 +202,7 @@ dummyRQ userId server connId rcvId =
       queueMode = Just QMMessaging,
       shortLink = Nothing,
       status = New,
-      dbQueueId = DBQueueId 0,
+      dbQueueId = DBEntityId 0,
       primary = True,
       dbReplaceQueueId = Nothing,
       rcvSwchStatus = Nothing,

tests/CoreTests/UtilTests.hs

@@ -8,8 +8,9 @@ import Control.Monad.Except
 import Control.Monad.IO.Class
 import Data.IORef
 import Simplex.Messaging.Util
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
 import qualified UnliftIO.Exception as UE
+import Util
 
 utilTests :: Spec
 utilTests = do

tests/CoreTests/VersionRangeTests.hs

@@ -11,9 +11,10 @@ import GHC.Generics (Generic)
 import Generic.Random (genericArbitraryU)
 import Simplex.Messaging.Version
 import Simplex.Messaging.Version.Internal
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
 import Test.Hspec.QuickCheck (modifyMaxSuccess)
 import Test.QuickCheck
+import Util
 
 data V = V1 | V2 | V3 | V4 | V5 deriving (Eq, Enum, Ord, Generic, Show)
 

tests/FileDescriptionTests.hs

@@ -16,7 +16,8 @@ import Simplex.Messaging.Encoding.String (StrEncoding (..))
 import Simplex.Messaging.Protocol (EntityId (..))
 import Simplex.Messaging.ServiceScheme (ServiceScheme (..))
 import System.Directory (removeFile)
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
+import Util
 
 fileDescriptionTests :: Spec
 fileDescriptionTests = do

tests/NtfClient.hs

@@ -55,7 +55,7 @@ import Simplex.Messaging.Transport.Client
 import Simplex.Messaging.Transport.HTTP2 (HTTP2Body (..), http2TLSParams)
 import Simplex.Messaging.Transport.HTTP2.Server
 import Simplex.Messaging.Transport.Server
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
 import UnliftIO.Async
 import UnliftIO.Concurrent
 import qualified UnliftIO.Exception as E
@@ -83,7 +83,7 @@ ntfTestPrometheusMetricsFile :: FilePath
 ntfTestPrometheusMetricsFile = "tests/tmp/ntf-server-metrics.txt"
 
 ntfTestStoreDBOpts :: DBOpts
-ntfTestStoreDBOpts = 
+ntfTestStoreDBOpts =
   DBOpts
     { connstr = ntfTestServerDBConnstr,
       schema = "ntf_server",
@@ -99,10 +99,10 @@ ntfTestServerDBConnstr = "postgresql://ntf_test_server_user@/ntf_test_server_db"
 
 ntfTestServerDBConnectInfo :: ConnectInfo
 ntfTestServerDBConnectInfo =
-  defaultConnectInfo {
-    connectUser = "ntf_test_server_user",
-    connectDatabase = "ntf_test_server_db"
-  }
+  defaultConnectInfo
+    { connectUser = "ntf_test_server_user",
+      connectDatabase = "ntf_test_server_db"
+    }
 
 ntfTestDBCfg :: PostgresStoreCfg
 ntfTestDBCfg =
@@ -134,7 +134,6 @@ ntfServerCfg =
       subIdBytes = 24,
       regCodeBytes = 32,
       clientQSize = 2,
-      subQSize = 2,
       pushQSize = 2,
       smpAgentCfg = defaultSMPClientAgentConfig {persistErrorInterval = 0},
       apnsConfig =
@@ -175,7 +174,7 @@ ntfServerCfgVPrev =
     smpCfg' = smpCfg smpAgentCfg'
     serverVRange' = serverVRange smpCfg'
 
-withNtfServerThreadOn :: HasCallStack => ATransport -> ServiceName -> PostgresStoreCfg -> (HasCallStack => ThreadId -> IO a) -> IO a
+withNtfServerThreadOn :: HasCallStack => ASrvTransport -> ServiceName -> PostgresStoreCfg -> (HasCallStack => ThreadId -> IO a) -> IO a
 withNtfServerThreadOn t port' dbStoreConfig =
   withNtfServerCfg ntfServerCfg {transports = [(port', t, False)], dbStoreConfig}
 
@@ -188,10 +187,10 @@ withNtfServerCfg cfg@NtfServerConfig {transports} =
         (\started -> runNtfServerBlocking started cfg)
         (pure ())
 
-withNtfServerOn :: HasCallStack => ATransport -> ServiceName -> PostgresStoreCfg -> (HasCallStack => IO a) -> IO a
+withNtfServerOn :: HasCallStack => ASrvTransport -> ServiceName -> PostgresStoreCfg -> (HasCallStack => IO a) -> IO a
 withNtfServerOn t port' dbStoreConfig = withNtfServerThreadOn t port' dbStoreConfig . const
 
-withNtfServer :: HasCallStack => ATransport -> (HasCallStack => IO a) -> IO a
+withNtfServer :: HasCallStack => ASrvTransport -> (HasCallStack => IO a) -> IO a
 withNtfServer t = withNtfServerOn t ntfTestPort ntfTestDBCfg
 
 runNtfTest :: forall c a. Transport c => (THandleNTF c 'TClient -> IO a) -> IO a
@@ -200,7 +199,7 @@ runNtfTest test = withNtfServer (transport @c) $ testNtfClient test
 ntfServerTest ::
   forall c smp.
   (Transport c, Encoding smp) =>
-  TProxy c ->
+  TProxy c 'TServer ->
   (Maybe TransmissionAuth, ByteString, ByteString, smp) ->
   IO (Maybe TransmissionAuth, ByteString, ByteString, NtfResponse)
 ntfServerTest _ t = runNtfTest $ \h -> tPut' h t >> tGet' h
@@ -214,7 +213,7 @@ ntfServerTest _ t = runNtfTest $ \h -> tPut' h t >> tGet' h
       [(Nothing, _, (CorrId corrId, EntityId qId, Right cmd))] <- tGet h
       pure (Nothing, corrId, qId, cmd)
 
-ntfTest :: Transport c => TProxy c -> (THandleNTF c 'TClient -> IO ()) -> Expectation
+ntfTest :: Transport c => TProxy c 'TServer -> (THandleNTF c 'TClient -> IO ()) -> Expectation
 ntfTest _ test' = runNtfTest test' `shouldReturn` ()
 
 data APNSMockRequest = APNSMockRequest

tests/NtfServerTests.hs

@@ -43,17 +43,18 @@ import Simplex.Messaging.Notifications.Transport (THandleNTF)
 import Simplex.Messaging.Parsers (parse, parseAll)
 import Simplex.Messaging.Protocol hiding (notification)
 import Simplex.Messaging.Transport
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
 import UnliftIO.STM
+import Util
 
-ntfServerTests :: ATransport -> Spec
+ntfServerTests :: ASrvTransport -> Spec
 ntfServerTests t = do
   describe "Notifications server protocol syntax" $ ntfSyntaxTests t
   describe "Notification subscriptions (NKEY)" $ testNotificationSubscription t createNtfQueueNKEY
   -- describe "Notification subscriptions (NEW with ntf creds)" $ testNotificationSubscription t createNtfQueueNEW
   describe "Retried notification subscription" $ testRetriedNtfSubscription t
 
-ntfSyntaxTests :: ATransport -> Spec
+ntfSyntaxTests :: ASrvTransport -> Spec
 ntfSyntaxTests (ATransport t) = do
   it "unknown command" $ ("", "abcd", "1234", ('H', 'E', 'L', 'L', 'O')) >#> ("", "abcd", "1234", NRErr $ CMD UNKNOWN)
   describe "NEW" $ do
@@ -96,7 +97,7 @@ v .-> key =
   let J.Object o = v
    in U.decodeLenient . encodeUtf8 <$> JT.parseEither (J..: key) o
 
-testNotificationSubscription :: ATransport -> CreateQueueFunc -> Spec
+testNotificationSubscription :: ASrvTransport -> CreateQueueFunc -> Spec
 testNotificationSubscription (ATransport t) createQueue =
   it "should create notification subscription and notify when message is received" $ do
     g <- C.newRandom
@@ -179,7 +180,7 @@ testNotificationSubscription (ATransport t) createQueue =
           smpServer3 `shouldBe` srv
           notifierId3 `shouldBe` nId
 
-testRetriedNtfSubscription :: ATransport -> Spec
+testRetriedNtfSubscription :: ASrvTransport -> Spec
 testRetriedNtfSubscription (ATransport t) =
   it "should allow retrying to create notification subscription with the same token and key" $ do
     g <- C.newRandom
@@ -243,7 +244,7 @@ registerToken nh apns token = do
         let Right verification = nd .-> "verification"
             Right nonce = C.cbNonce <$> nd .-> "nonce"
             Right pt = C.cbDecrypt dhSecret nonce verification
-          in NtfRegCode pt
+         in NtfRegCode pt
   let code = decryptCode ntfData
   pure (tknKey, dhSecret, tId, code)
 

tests/PostgresSchemaDump.hs

@@ -17,7 +17,8 @@ import Simplex.Messaging.Util (ifM, whenM)
 import System.Directory (doesFileExist, removeFile)
 import System.Environment (lookupEnv)
 import System.Process (readCreateProcess, shell)
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
+import Util
 
 testSchemaPath :: FilePath
 testSchemaPath = "tests/tmp/test_schema.sql"

tests/RemoteControl.hs

@@ -19,9 +19,10 @@ import qualified Simplex.RemoteControl.Client as RC
 import Simplex.RemoteControl.Discovery (mkLastLocalHost, preferAddress)
 import Simplex.RemoteControl.Invitation (RCSignedInvitation, verifySignedInvitation)
 import Simplex.RemoteControl.Types
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
 import UnliftIO
 import UnliftIO.Concurrent
+import Util
 
 remoteControlTests :: Spec
 remoteControlTests = do

tests/SMPClient.hs

@@ -14,7 +14,6 @@
 
 module SMPClient where
 
-import Control.Logger.Simple (LogLevel (..))
 import Control.Monad.Except (runExceptT)
 import Data.ByteString.Char8 (ByteString)
 import Data.List.NonEmpty (NonEmpty)
@@ -38,7 +37,7 @@ import Simplex.Messaging.Version
 import Simplex.Messaging.Version.Internal
 import System.Environment (lookupEnv)
 import System.Info (os)
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
 import UnliftIO.Concurrent
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM (TMVar, atomically, newEmptyTMVarIO, putTMVar, takeTMVar)
@@ -82,7 +81,7 @@ testStoreLogFile2 :: FilePath
 testStoreLogFile2 = "tests/tmp/smp-server-store.log.2"
 
 testStoreDBOpts :: DBOpts
-testStoreDBOpts = 
+testStoreDBOpts =
   DBOpts
     { connstr = testServerDBConnstr,
       schema = "smp_server",
@@ -176,7 +175,7 @@ journalCfg :: ServerConfig -> FilePath -> FilePath -> ServerConfig
 journalCfg cfg' storeLogFile storeMsgsPath = cfg' {serverStoreCfg = ASSCfg SQSMemory SMSJournal SSCMemoryJournal {storeLogFile, storeMsgsPath}}
 
 journalCfgDB :: ServerConfig -> DBOpts -> FilePath -> ServerConfig
-journalCfgDB cfg' dbOpts storeMsgsPath' = 
+journalCfgDB cfg' dbOpts storeMsgsPath' =
   let storeCfg = PostgresStoreCfg {dbOpts, dbStoreLogPath = Nothing, confirmMigrations = MCYesUp, deletedTTL = 86400}
    in cfg' {serverStoreCfg = ASSCfg SQSPostgres SMSJournal SSCDatabaseJournal {storeCfg, storeMsgsPath'}}
 
@@ -191,7 +190,7 @@ cfgMS msType =
       maxJournalStateLines = 2,
       queueIdBytes = 24,
       msgIdBytes = 24,
-      serverStoreCfg = serverStoreConfig msType,
+      serverStoreCfg = testServerStoreConfig msType,
       storeNtfsFile = Nothing,
       allowNewQueues = True,
       newQueueBasicAuth = Nothing,
@@ -228,10 +227,10 @@ cfgMS msType =
     }
 
 defaultStartOptions :: StartOptions
-defaultStartOptions = StartOptions {maintenance = False, compactLog = False, logLevel = LogError, skipWarnings = False, confirmMigrations = MCYesUp} 
+defaultStartOptions = StartOptions {maintenance = False, compactLog = False, logLevel = testLogLevel, skipWarnings = False, confirmMigrations = MCYesUp}
 
-serverStoreConfig :: AStoreType -> AServerStoreCfg
-serverStoreConfig = serverStoreConfig_ False
+testServerStoreConfig :: AStoreType -> AServerStoreCfg
+testServerStoreConfig = serverStoreConfig_ False
 
 serverStoreConfig_ :: Bool -> AStoreType -> AServerStoreCfg
 serverStoreConfig_ useDbStoreLog = \case
@@ -282,20 +281,20 @@ proxyCfgJ2QS = \case
 proxyVRangeV8 :: VersionRangeSMP
 proxyVRangeV8 = mkVersionRange minServerSMPRelayVersion sendingProxySMPVersion
 
-withSmpServerStoreMsgLogOn :: HasCallStack => (ATransport, AStoreType) -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
+withSmpServerStoreMsgLogOn :: HasCallStack => (ASrvTransport, AStoreType) -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
 withSmpServerStoreMsgLogOn (t, msType) =
   withSmpServerConfigOn t (cfgMS msType) {storeNtfsFile = Just testStoreNtfsFile, serverStatsBackupFile = Just testServerStatsBackupFile}
 
-withSmpServerStoreLogOn :: HasCallStack => (ATransport, AStoreType) -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
+withSmpServerStoreLogOn :: HasCallStack => (ASrvTransport, AStoreType) -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
 withSmpServerStoreLogOn (t, msType) = withSmpServerConfigOn t (cfgMS msType) {serverStatsBackupFile = Just testServerStatsBackupFile}
 
-withSmpServerConfigOn :: HasCallStack => ATransport -> ServerConfig -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
+withSmpServerConfigOn :: HasCallStack => ASrvTransport -> ServerConfig -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
 withSmpServerConfigOn t cfg' port' =
   serverBracket
     (\started -> runSMPServerBlocking started cfg' {transports = [(port', t, False)]} Nothing)
     (threadDelay 10000)
 
-withSmpServerThreadOn :: HasCallStack => (ATransport, AStoreType) -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
+withSmpServerThreadOn :: HasCallStack => (ASrvTransport, AStoreType) -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
 withSmpServerThreadOn (t, msType) = withSmpServerConfigOn t (cfgMS msType)
 
 serverBracket :: HasCallStack => (TMVar Bool -> IO ()) -> IO () -> (HasCallStack => ThreadId -> IO a) -> IO a
@@ -316,19 +315,19 @@ serverBracket process afterProcess f = do
         Nothing -> error $ "server did not " <> s
         _ -> pure ()
 
-withSmpServerOn :: HasCallStack => (ATransport, AStoreType) -> ServiceName -> IO a -> IO a
+withSmpServerOn :: HasCallStack => (ASrvTransport, AStoreType) -> ServiceName -> IO a -> IO a
 withSmpServerOn ps port' = withSmpServerThreadOn ps port' . const
 
-withSmpServer :: HasCallStack => (ATransport, AStoreType) -> IO a -> IO a
+withSmpServer :: HasCallStack => (ASrvTransport, AStoreType) -> IO a -> IO a
 withSmpServer ps = withSmpServerOn ps testPort
 
-withSmpServerProxy :: HasCallStack => (ATransport, AStoreType) -> IO a -> IO a
+withSmpServerProxy :: HasCallStack => (ASrvTransport, AStoreType) -> IO a -> IO a
 withSmpServerProxy (t, msType) = withSmpServerConfigOn t (proxyCfgMS msType) testPort . const
 
-withSmpServers2 :: HasCallStack => (ATransport, AStoreType) -> IO a -> IO a
+withSmpServers2 :: HasCallStack => (ASrvTransport, AStoreType) -> IO a -> IO a
 withSmpServers2 ps@(t, ASType qs _ms) = withSmpServer ps . withSmpServerConfigOn t (cfgJ2QS qs) testPort2 . const
 
-withSmpServersProxy2 :: HasCallStack => (ATransport, AStoreType) -> IO a -> IO a
+withSmpServersProxy2 :: HasCallStack => (ASrvTransport, AStoreType) -> IO a -> IO a
 withSmpServersProxy2 ps@(t, ASType qs _ms) = withSmpServerProxy ps . withSmpServerConfigOn t (proxyCfgJ2QS qs) testPort2 . const
 
 runSmpTest :: forall c a. (HasCallStack, Transport c) => AStoreType -> (HasCallStack => THandleSMP c 'TClient -> IO a) -> IO a
@@ -347,7 +346,7 @@ runSmpTestNCfg srvCfg clntVR nClients test = withSmpServerConfigOn (transport @c
 smpServerTest ::
   forall c smp.
   (Transport c, Encoding smp) =>
-  TProxy c ->
+  TProxy c 'TServer ->
   (Maybe TransmissionAuth, ByteString, ByteString, smp) ->
   IO (Maybe TransmissionAuth, ByteString, ByteString, BrokerMsg)
 smpServerTest _ t = runSmpTest (ASType SQSMemory SMSJournal) $ \h -> tPut' h t >> tGet' h
@@ -361,36 +360,36 @@ smpServerTest _ t = runSmpTest (ASType SQSMemory SMSJournal) $ \h -> tPut' h t >
       [(Nothing, _, (CorrId corrId, EntityId qId, Right cmd))] <- tGet h
       pure (Nothing, corrId, qId, cmd)
 
-smpTest :: (HasCallStack, Transport c) => TProxy c -> AStoreType -> (HasCallStack => THandleSMP c 'TClient -> IO ()) -> Expectation
+smpTest :: (HasCallStack, Transport c) => TProxy c 'TServer -> AStoreType -> (HasCallStack => THandleSMP c 'TClient -> IO ()) -> Expectation
 smpTest _ msType test' = runSmpTest msType test' `shouldReturn` ()
 
-smpTest' :: forall c. (HasCallStack, Transport c) => TProxy c -> (HasCallStack => THandleSMP c 'TClient -> IO ()) -> Expectation
+smpTest' :: forall c. (HasCallStack, Transport c) => TProxy c 'TServer -> (HasCallStack => THandleSMP c 'TClient -> IO ()) -> Expectation
 smpTest' = (`smpTest` ASType SQSMemory SMSJournal)
 
 smpTestN :: (HasCallStack, Transport c) => AStoreType -> Int -> (HasCallStack => [THandleSMP c 'TClient] -> IO ()) -> Expectation
 smpTestN msType n test' = runSmpTestN msType n test' `shouldReturn` ()
 
-smpTest2' :: forall c. (HasCallStack, Transport c) => TProxy c -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
+smpTest2' :: forall c. (HasCallStack, Transport c) => TProxy c 'TServer -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
 smpTest2' = (`smpTest2` ASType SQSMemory SMSJournal)
 
-smpTest2 :: forall c. (HasCallStack, Transport c) => TProxy c -> AStoreType -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
+smpTest2 :: forall c. (HasCallStack, Transport c) => TProxy c 'TServer -> AStoreType -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
 smpTest2 t msType = smpTest2Cfg (cfgMS msType) supportedClientSMPRelayVRange t
 
-smpTest2Cfg :: forall c. (HasCallStack, Transport c) => ServerConfig -> VersionRangeSMP -> TProxy c -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
+smpTest2Cfg :: forall c. (HasCallStack, Transport c) => ServerConfig -> VersionRangeSMP -> TProxy c 'TServer -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
 smpTest2Cfg srvCfg clntVR _ test' = runSmpTestNCfg srvCfg clntVR 2 _test `shouldReturn` ()
   where
     _test :: HasCallStack => [THandleSMP c 'TClient] -> IO ()
     _test [h1, h2] = test' h1 h2
     _test _ = error "expected 2 handles"
 
-smpTest3 :: forall c. (HasCallStack, Transport c) => TProxy c -> AStoreType -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
+smpTest3 :: forall c. (HasCallStack, Transport c) => TProxy c 'TServer -> AStoreType -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
 smpTest3 _ msType test' = smpTestN msType 3 _test
   where
     _test :: HasCallStack => [THandleSMP c 'TClient] -> IO ()
     _test [h1, h2, h3] = test' h1 h2 h3
     _test _ = error "expected 3 handles"
 
-smpTest4 :: forall c. (HasCallStack, Transport c) => TProxy c -> AStoreType -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
+smpTest4 :: forall c. (HasCallStack, Transport c) => TProxy c 'TServer -> AStoreType -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
 smpTest4 _ msType test' = smpTestN msType 4 _test
   where
     _test :: HasCallStack => [THandleSMP c 'TClient] -> IO ()

tests/SMPProxyTests.hs

@@ -45,7 +45,7 @@ import Simplex.Messaging.Util (bshow, tshow)
 import Simplex.Messaging.Version (mkVersionRange)
 import System.FilePath (splitExtensions)
 import System.Random (randomRIO)
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
 import UnliftIO
 import Util
 #if defined(dbPostgres)

tests/ServerTests.hs

@@ -17,7 +17,7 @@ module ServerTests where
 
 import Control.Concurrent (ThreadId, killThread, threadDelay)
 import Control.Concurrent.STM
-import Control.Exception (SomeException, try, throwIO)
+import Control.Exception (SomeException, throwIO, try)
 import Control.Monad
 import Control.Monad.IO.Class
 import CoreTests.MsgStoreTests (testJournalStoreCfg)
@@ -39,7 +39,7 @@ import Simplex.Messaging.Server (exportMessages)
 import Simplex.Messaging.Server.Env.STM (AServerStoreCfg (..), AStoreType (..), ServerConfig (..), ServerStoreCfg (..), readWriteQueueStore)
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.MsgStore.Journal (JournalStoreConfig (..), QStoreCfg (..))
-import Simplex.Messaging.Server.MsgStore.Types (MsgStoreClass (..), SQSType (..), SMSType (..), newMsgStore)
+import Simplex.Messaging.Server.MsgStore.Types (MsgStoreClass (..), SMSType (..), SQSType (..), newMsgStore)
 import Simplex.Messaging.Server.Stats (PeriodStatsData (..), ServerStatsData (..))
 import Simplex.Messaging.Server.StoreLog (StoreLogRecord (..), closeStoreLog)
 import Simplex.Messaging.Transport
@@ -50,10 +50,10 @@ import System.IO (IOMode (..), withFile)
 import System.TimeIt (timeItT)
 import System.Timeout
 import Test.HUnit
-import Test.Hspec
-import Util (removeFileIfExists)
+import Test.Hspec hiding (fit, it)
+import Util
 
-serverTests :: SpecWith (ATransport, AStoreType)
+serverTests :: SpecWith (ASrvTransport, AStoreType)
 serverTests = do
   describe "SMP queues" $ do
     describe "NEW and KEY commands, SEND messages" testCreateSecure
@@ -147,7 +147,7 @@ decryptMsgV3 dhShared nonce body =
     Right ClientRcvMsgQuota {} -> Left "ClientRcvMsgQuota"
     Left e -> Left e
 
-testCreateSecure :: SpecWith (ATransport, AStoreType)
+testCreateSecure :: SpecWith (ASrvTransport, AStoreType)
 testCreateSecure =
   it "should create (NEW) and secure (KEY) queue" $ \(ATransport t, msType) ->
     smpTest2 t msType $ \r s -> do
@@ -212,7 +212,7 @@ testCreateSecure =
       Resp "bcda" _ (ERR LARGE_MSG) <- signSendRecv s sKey ("bcda", sId, _SEND biggerMessage)
       pure ()
 
-testCreateSndSecure :: SpecWith (ATransport, AStoreType)
+testCreateSndSecure :: SpecWith (ASrvTransport, AStoreType)
 testCreateSndSecure =
   it "should create (NEW) and secure (SKEY) queue by sender" $ \(ATransport t, msType) ->
     smpTest2 t msType $ \r s -> do
@@ -259,7 +259,7 @@ testCreateSndSecure =
       Resp "bcda" _ (ERR LARGE_MSG) <- signSendRecv s sKey ("bcda", sId, _SEND biggerMessage)
       pure ()
 
-testSndSecureProhibited :: SpecWith (ATransport, AStoreType)
+testSndSecureProhibited :: SpecWith (ASrvTransport, AStoreType)
 testSndSecureProhibited =
   it "should create (NEW) without allowing sndSecure and fail to and secure queue by sender (SKEY)" $ \(ATransport t, msType) ->
     smpTest2 t msType $ \r s -> do
@@ -274,7 +274,7 @@ testSndSecureProhibited =
       (sId2, sId) #== "secures queue, same queue ID in response"
       (err, ERR AUTH) #== "rejects SKEY when not allowed in NEW command"
 
-testCreateUpdateKeys :: SpecWith (ATransport, AStoreType)
+testCreateUpdateKeys :: SpecWith (ASrvTransport, AStoreType)
 testCreateUpdateKeys =
   it "should create (NEW) and updated recipient keys (RKEY)" $ \(ATransport t, msType) ->
     smpTest t msType $ \h -> do
@@ -306,7 +306,7 @@ testCreateUpdateKeys =
       Resp "11" _ (INFO _) <- signSendRecv h rKey' ("11", rId, QUE)
       pure ()
 
-testCreateDelete :: SpecWith (ATransport, AStoreType)
+testCreateDelete :: SpecWith (ASrvTransport, AStoreType)
 testCreateDelete =
   it "should create (NEW), suspend (OFF) and delete (DEL) queue" $ \(ATransport t, msType) ->
     smpTest2 t msType $ \rh sh -> do
@@ -377,7 +377,7 @@ testCreateDelete =
       Resp "cdab" _ err10 <- signSendRecv rh rKey ("cdab", rId, SUB)
       (err10, ERR AUTH) #== "rejects SUB when deleted"
 
-stressTest :: SpecWith (ATransport, AStoreType)
+stressTest :: SpecWith (ASrvTransport, AStoreType)
 stressTest =
   it "should create many queues, disconnect and re-connect" $ \(ATransport t, msType) ->
     smpTest3 t msType $ \h1 h2 h3 -> do
@@ -395,9 +395,9 @@ stressTest =
       closeConnection $ connection h2
       subscribeQueues h3
 
-testAllowNewQueues :: SpecWith (ATransport, AStoreType)
+testAllowNewQueues :: SpecWith (ASrvTransport, AStoreType)
 testAllowNewQueues =
-  it "should prohibit creating new queues with allowNewQueues = False" $ \(ATransport (t :: TProxy c), msType) ->
+  it "should prohibit creating new queues with allowNewQueues = False" $ \(ATransport (t :: TProxy c 'TServer), msType) ->
     withSmpServerConfigOn (ATransport t) (cfgMS msType) {allowNewQueues = False} testPort $ \_ ->
       testSMPClient @c $ \h -> do
         g <- C.newRandom
@@ -406,7 +406,7 @@ testAllowNewQueues =
         Resp "abcd" NoEntity (ERR AUTH) <- signSendRecv h rKey ("abcd", NoEntity, New rPub dhPub)
         pure ()
 
-testDuplex :: SpecWith (ATransport, AStoreType)
+testDuplex :: SpecWith (ASrvTransport, AStoreType)
 testDuplex =
   it "should create 2 simplex connections and exchange messages" $ \(ATransport t, msType) ->
     smpTest2 t msType $ \alice bob -> do
@@ -461,7 +461,7 @@ testDuplex =
       Resp "bcda" _ OK <- signSendRecv bob brKey ("bcda", bRcv, ACK mId5)
       (bDec mId5 msg5, Right "how are you bob") #== "message received from alice"
 
-testSwitchSub :: SpecWith (ATransport, AStoreType)
+testSwitchSub :: SpecWith (ASrvTransport, AStoreType)
 testSwitchSub =
   it "should create simplex connections and switch subscription to another TCP connection" $ \(ATransport t, msType) ->
     smpTest3 t msType $ \rh1 rh2 sh -> do
@@ -506,9 +506,9 @@ testSwitchSub =
         Nothing -> return ()
         Just _ -> error "nothing else is delivered to the 1st TCP connection"
 
-testGetCommand :: SpecWith (ATransport, AStoreType)
+testGetCommand :: SpecWith (ASrvTransport, AStoreType)
 testGetCommand =
-  it "should retrieve messages from the queue using GET command" $ \(ATransport (t :: TProxy c), msType) -> do
+  it "should retrieve messages from the queue using GET command" $ \(ATransport (t :: TProxy c 'TServer), msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
     smpTest t msType $ \sh -> do
@@ -525,7 +525,7 @@ testGetCommand =
         Resp "4" _ OK <- signSendRecv rh rKey ("4", rId, GET)
         pure ()
 
-testGetSubCommands :: SpecWith (ATransport, AStoreType)
+testGetSubCommands :: SpecWith (ASrvTransport, AStoreType)
 testGetSubCommands =
   it "should retrieve messages with GET and receive with SUB, only one ACK would work" $ \(ATransport t, msType) -> do
     g <- C.newRandom
@@ -575,9 +575,9 @@ testGetSubCommands =
       Resp "12" _ OK <- signSendRecv rh2 rKey ("12", rId, GET)
       pure ()
 
-testExceedQueueQuota :: SpecWith (ATransport, AStoreType)
+testExceedQueueQuota :: SpecWith (ASrvTransport, AStoreType)
 testExceedQueueQuota =
-  it "should reply with ERR QUOTA to sender and send QUOTA message to the recipient" $ \(ATransport (t :: TProxy c), msType) -> do
+  it "should reply with ERR QUOTA to sender and send QUOTA message to the recipient" $ \(ATransport (t :: TProxy c 'TServer), msType) -> do
     withSmpServerConfigOn (ATransport t) (cfgMS msType) {msgQueueQuota = 2} testPort $ \_ ->
       testSMPClient @c $ \sh -> testSMPClient @c $ \rh -> do
         g <- C.newRandom
@@ -602,7 +602,7 @@ testExceedQueueQuota =
         Resp "10" _ OK <- signSendRecv rh rKey ("10", rId, ACK mId4)
         pure ()
 
-testWithStoreLog :: SpecWith (ATransport, AStoreType)
+testWithStoreLog :: SpecWith (ASrvTransport, AStoreType)
 testWithStoreLog =
   it "should store simplex queues to log and restore them after server restart" $ \(at@(ATransport t), msType) -> do
     g <- C.newRandom
@@ -678,16 +678,16 @@ testWithStoreLog =
     logSize testStoreLogFile `shouldReturn` (if compacting then 1 else 6)
     removeFile testStoreLogFile
   where
-    runTest :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> ThreadId -> Expectation
+    runTest :: Transport c => TProxy c 'TServer -> (THandleSMP c 'TClient -> IO ()) -> ThreadId -> Expectation
     runTest _ test' server = do
       testSMPClient test' `shouldReturn` ()
       killThread server
 
-    runClient :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> Expectation
+    runClient :: Transport c => TProxy c 'TServer -> (THandleSMP c 'TClient -> IO ()) -> Expectation
     runClient _ test' = testSMPClient test' `shouldReturn` ()
 
 serverStoreLogCfg :: AStoreType -> (ServerConfig, Bool)
-serverStoreLogCfg msType = 
+serverStoreLogCfg msType =
   let serverStoreCfg = serverStoreConfig_ True msType
       cfg' = (cfgMS msType) {serverStoreCfg, storeNtfsFile = Just testStoreNtfsFile, serverStatsBackupFile = Just testServerStatsBackupFile}
       compacting = case msType of
@@ -705,7 +705,7 @@ logSize f = go (10 :: Int)
           | n > 0 -> threadDelay 100000 >> go (n - 1)
           | otherwise -> throwIO e
 
-testRestoreMessages :: SpecWith (ATransport, AStoreType)
+testRestoreMessages :: SpecWith (ASrvTransport, AStoreType)
 testRestoreMessages =
   it "should store messages on exit and restore on start" $ \(at@(ATransport t), msType) -> do
     removeFileIfExists testStoreLogFile
@@ -783,12 +783,12 @@ testRestoreMessages =
     whenM (doesDirectoryExist testStoreMsgsDir) $ removeDirectoryRecursive testStoreMsgsDir
     removeFile testServerStatsBackupFile
   where
-    runTest :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> ThreadId -> Expectation
+    runTest :: Transport c => TProxy c 'TServer -> (THandleSMP c 'TClient -> IO ()) -> ThreadId -> Expectation
     runTest _ test' server = do
       testSMPClient test' `shouldReturn` ()
       killThread server
 
-    runClient :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> Expectation
+    runClient :: Transport c => TProxy c 'TServer -> (THandleSMP c 'TClient -> IO ()) -> Expectation
     runClient _ test' = testSMPClient test' `shouldReturn` ()
 
 checkStats :: ServerStatsData -> [RecipientId] -> Int -> Int -> Expectation
@@ -807,7 +807,7 @@ checkStats s qs sent received = do
   IS.toList _week `shouldBe` map (hash . unEntityId) qs
   IS.toList _month `shouldBe` map (hash . unEntityId) qs
 
-testRestoreExpireMessages :: SpecWith (ATransport, AStoreType)
+testRestoreExpireMessages :: SpecWith (ASrvTransport, AStoreType)
 testRestoreExpireMessages =
   it "should store messages on exit and restore on start (old / v2)" $ \(at@(ATransport t), msType) -> do
     g <- C.newRandom
@@ -869,15 +869,15 @@ testRestoreExpireMessages =
           removeFileIfExists testStoreMsgsFile
           exportMessages False ms testStoreMsgsFile False
           closeMsgStore ms
-    runTest :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> ThreadId -> Expectation
+    runTest :: Transport c => TProxy c 'TServer -> (THandleSMP c 'TClient -> IO ()) -> ThreadId -> Expectation
     runTest _ test' server = do
       testSMPClient test' `shouldReturn` ()
       killThread server
 
-    runClient :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> Expectation
+    runClient :: Transport c => TProxy c 'TServer -> (THandleSMP c 'TClient -> IO ()) -> Expectation
     runClient _ test' = testSMPClient test' `shouldReturn` ()
 
-testPrometheusMetrics :: SpecWith (ATransport, AStoreType)
+testPrometheusMetrics :: SpecWith (ASrvTransport, AStoreType)
 testPrometheusMetrics =
   it "should save Prometheus metrics" $ \(at, msType) -> do
     let cfg' = (cfgMS msType) {prometheusInterval = Just 1}
@@ -895,7 +895,7 @@ createAndSecureQueue h sPub = do
   (rId', rId) #== "same queue ID"
   pure (sId, rId, rKey, dhShared)
 
-testTiming :: SpecWith (ATransport, AStoreType)
+testTiming :: SpecWith (ASrvTransport, AStoreType)
 testTiming =
   describe "should have similar time for auth error, whether queue exists or not, for all key types" $
     forM_ timingTests $ \tst ->
@@ -918,7 +918,9 @@ testTiming =
         (C.AuthAlg C.SX25519, C.AuthAlg C.SX25519, 200) -- correct key type
       ]
     timeRepeat n = fmap fst . timeItT . forM_ (replicate n ()) . const
-    similarTime t1 t2 = abs (t2 / t1 - 1) < 0.30 -- normally the difference between "no queue" and "wrong key" is less than 5%
+    similarTime t1 t2
+      | t1 <= t2 = abs (1 - t1 / t2) < 0.35 -- normally the difference between "no queue" and "wrong key" is less than 5%
+      | otherwise = similarTime t2 t1
     testSameTiming :: forall c. Transport c => THandleSMP c 'TClient -> THandleSMP c 'TClient -> (C.AuthAlg, C.AuthAlg, Int) -> Expectation
     testSameTiming rh sh (C.AuthAlg goodKeyAlg, C.AuthAlg badKeyAlg, n) = do
       g <- C.newRandom
@@ -965,7 +967,7 @@ testTiming =
             ]
           ok `shouldBe` True
 
-testMessageNotifications :: SpecWith (ATransport, AStoreType)
+testMessageNotifications :: SpecWith (ASrvTransport, AStoreType)
 testMessageNotifications =
   it "should create simplex connection, subscribe notifier and deliver notifications" $ \(ATransport t, msType) -> do
     g <- C.newRandom
@@ -1005,9 +1007,9 @@ testMessageNotifications =
         Nothing -> pure ()
         Just _ -> error "nothing else should be delivered to the 2nd notifier's TCP connection"
 
-testMsgExpireOnSend :: SpecWith (ATransport, AStoreType)
+testMsgExpireOnSend :: SpecWith (ASrvTransport, AStoreType)
 testMsgExpireOnSend =
-  it "should expire messages that are not received before messageTTL on SEND" $ \(ATransport (t :: TProxy c), msType) -> do
+  it "should expire messages that are not received before messageTTL on SEND" $ \(ATransport (t :: TProxy c 'TServer), msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
     let cfg' = (cfgMS msType) {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 10000}}
@@ -1025,10 +1027,10 @@ testMsgExpireOnSend =
             Nothing -> return ()
             Just _ -> error "nothing else should be delivered"
 
-testMsgExpireOnInterval :: SpecWith (ATransport, AStoreType)
+testMsgExpireOnInterval :: SpecWith (ASrvTransport, AStoreType)
 testMsgExpireOnInterval =
   -- fails on ubuntu
-  xit' "should expire messages that are not received before messageTTL after expiry interval" $ \(ATransport (t :: TProxy c), msType) -> do
+  xit' "should expire messages that are not received before messageTTL after expiry interval" $ \(ATransport (t :: TProxy c 'TServer), msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
     let cfg' = (cfgMS msType) {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}, idleQueueInterval = 1}
@@ -1045,9 +1047,9 @@ testMsgExpireOnInterval =
             Nothing -> return ()
             Just _ -> error "nothing should be delivered"
 
-testMsgNOTExpireOnInterval :: SpecWith (ATransport, AStoreType)
+testMsgNOTExpireOnInterval :: SpecWith (ASrvTransport, AStoreType)
 testMsgNOTExpireOnInterval =
-  it "should block and unblock message queues" $ \(ATransport (t :: TProxy c), msType) -> do
+  it "should block and unblock message queues" $ \(ATransport (t :: TProxy c 'TServer), msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
     let cfg' = (cfgMS msType) {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 10000}}
@@ -1064,10 +1066,10 @@ testMsgNOTExpireOnInterval =
             Nothing -> return ()
             Just _ -> error "nothing else should be delivered"
 
-testBlockMessageQueue :: SpecWith (ATransport, AStoreType)
+testBlockMessageQueue :: SpecWith (ASrvTransport, AStoreType)
 testBlockMessageQueue =
   -- TODO [postgres]
-  xit "should return BLOCKED error when queue is blocked" $ \ps@(ATransport (t :: TProxy c), _) -> do
+  xit "should return BLOCKED error when queue is blocked" $ \ps@(ATransport (t :: TProxy c 'TServer), _) -> do
     g <- C.newRandom
     (rId, sId) <- withSmpServerStoreLogOn ps testPort $ runTest t $ \h -> do
       (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd448 g
@@ -1084,15 +1086,15 @@ testBlockMessageQueue =
       Resp "dabc" sId2 (ERR (BLOCKED (BlockingInfo BRContent))) <- signSendRecv h sKey ("dabc", sId, SKEY sPub)
       (sId2, sId) #== "same queue ID in response"
   where
-    runTest :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO a) -> ThreadId -> IO a
+    runTest :: Transport c => TProxy c 'TServer -> (THandleSMP c 'TClient -> IO a) -> ThreadId -> IO a
     runTest _ test' server = do
       a <- testSMPClient test'
       killThread server
       pure a
 
-testInvQueueLinkData :: SpecWith (ATransport, AStoreType)
-testInvQueueLinkData = 
-  it "create and access queue short link data for 1-time invitation"  $ \(ATransport t, msType) ->
+testInvQueueLinkData :: SpecWith (ASrvTransport, AStoreType)
+testInvQueueLinkData =
+  it "create and access queue short link data for 1-time invitation" $ \(ATransport t, msType) ->
     smpTest2 t msType $ \r s -> do
       g <- C.newRandom
       (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
@@ -1143,9 +1145,9 @@ testInvQueueLinkData =
       Resp "9" rId2 (ERR AUTH) <- signSendRecv r rKey ("9", rId, LDEL)
       rId2 `shouldBe` rId
 
-testContactQueueLinkData :: SpecWith (ATransport, AStoreType)
-testContactQueueLinkData = 
-  it "create and access queue short link data for contact address"  $ \(ATransport t, msType) ->
+testContactQueueLinkData :: SpecWith (ASrvTransport, AStoreType)
+testContactQueueLinkData =
+  it "create and access queue short link data for contact address" $ \(ATransport t, msType) ->
     smpTest2 t msType $ \r s -> do
       g <- C.newRandom
       (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
@@ -1222,7 +1224,7 @@ instance Eq C.ASignature where
     Just Refl -> s == s'
     _ -> False
 
-serverSyntaxTests :: ATransport -> Spec
+serverSyntaxTests :: ASrvTransport -> Spec
 serverSyntaxTests (ATransport t) = do
   it "unknown command" $ ("", "abcd", "1234", ('H', 'E', 'L', 'L', 'O')) >#> ("", "abcd", "1234", ERR $ CMD UNKNOWN)
   describe "NEW" $ do

tests/Test.hs

@@ -30,7 +30,8 @@ import Simplex.Messaging.Transport (TLS, Transport (..))
 -- import Simplex.Messaging.Transport.WebSockets (WS)
 import System.Directory (createDirectoryIfMissing, removeDirectoryRecursive)
 import System.Environment (setEnv)
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
+import Util
 import XFTPAgent
 import XFTPCLI
 import XFTPServerTests (xftpServerTests)
@@ -59,8 +60,7 @@ logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
 
 main :: IO ()
 main = do
-  -- TODO [ntfdb] running wiht LogWarn level shows potential issue "Queue count differs"
-  setLogLevel LogError -- LogInfo -- also change in SMPClient.hs in defaultStartOptions
+  setLogLevel testLogLevel
   withGlobalLogging logCfg $ do
     setEnv "APNS_KEY_ID" "H82WD9K9AQ"
     setEnv "APNS_KEY_FILE" "./tests/fixtures/AuthKey_H82WD9K9AQ.p8"

tests/Util.hs

@@ -1,12 +1,22 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeFamilies #-}
+
 module Util where
 
+import Control.Concurrent.Async
+import Control.Exception as E
+import Control.Logger.Simple
 import Control.Monad (replicateM, when)
 import Data.Either (partitionEithers)
 import Data.List (tails)
 import GHC.Conc (getNumCapabilities, getNumProcessors, setNumCapabilities)
 import System.Directory (doesFileExist, removeFile)
-import Test.Hspec
-import UnliftIO
+import System.Process (callCommand)
+import System.Timeout (timeout)
+import Test.Hspec hiding (fit, it)
+import qualified Test.Hspec as Hspec
+import Test.Hspec.Core.Spec (Example (..), Result (..), ResultStatus (..))
 
 skip :: String -> SpecWith a -> SpecWith a
 skip = before_ . pendingWith
@@ -32,3 +42,40 @@ removeFileIfExists :: FilePath -> IO ()
 removeFileIfExists filePath = do
   fileExists <- doesFileExist filePath
   when fileExists $ removeFile filePath
+
+newtype TestWrapper a = TestWrapper a
+
+-- TODO [ntfdb] running wiht LogWarn level shows potential issue "Queue count differs"
+testLogLevel :: LogLevel
+testLogLevel = LogError
+
+instance Example a => Example (TestWrapper a) where
+  type Arg (TestWrapper a) = Arg a
+  evaluateExample (TestWrapper action) params hooks state =
+    runTest `E.catches` [E.Handler onTestFailure, E.Handler onTestException]
+    where
+      tt = 120
+      runTest =
+        timeout (tt * 1000000) (evaluateExample action params hooks state) `finally` callCommand "sync" >>= \case
+          Just r -> pure r
+          Nothing -> throwIO $ userError $ "test timed out after " <> show tt <> " seconds"
+      onTestFailure :: ResultStatus -> IO Result
+      onTestFailure = \case
+        Failure loc_ reason -> do
+          putStrLn $ "Test failed: location " ++ show loc_ ++ ", reason: " ++ show reason
+          retryTest
+        r -> E.throwIO r
+      onTestException :: SomeException -> IO Result
+      onTestException e = do
+        putStrLn $ "Test exception: " ++ show e
+        retryTest
+      retryTest = do
+        putStrLn "Retrying with more logs..."
+        setLogLevel LogDebug
+        runTest `finally` setLogLevel testLogLevel -- change this to match log level in Test.hs
+
+it :: (HasCallStack, Example a) => String -> a -> SpecWith (Arg a)
+it label action = Hspec.it label (TestWrapper action)
+
+fit :: (HasCallStack, Example a) => String -> a -> SpecWith (Arg a)
+fit = fmap focus . it

tests/XFTPAgent.hs

@@ -42,9 +42,10 @@ import Simplex.Messaging.Transport (ALPN)
 import Simplex.Messaging.Util (tshow)
 import System.Directory (doesDirectoryExist, doesFileExist, getFileSize, listDirectory, removeFile)
 import System.FilePath ((</>))
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
 import UnliftIO
 import UnliftIO.Concurrent
+import Util
 import XFTPCLI
 import XFTPClient
 #if defined(dbPostgres)

tests/XFTPCLI.hs

@@ -9,7 +9,8 @@ import System.Directory (createDirectoryIfMissing, getFileSize, listDirectory, r
 import System.Environment (withArgs)
 import System.FilePath ((</>))
 import System.IO.Silently (capture_)
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
+import Util
 import XFTPClient (testXFTPServerStr, testXFTPServerStr2, withXFTPServer, withXFTPServer2, xftpServerFiles, xftpServerFiles2)
 
 xftpCLITests :: Spec

tests/XFTPClient.hs

@@ -19,7 +19,7 @@ import Simplex.FileTransfer.Transport (supportedFileServerVRange, supportedXFTPh
 import Simplex.Messaging.Protocol (XFTPServer)
 import Simplex.Messaging.Transport (ALPN)
 import Simplex.Messaging.Transport.Server
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
 
 xftpTest :: HasCallStack => (HasCallStack => XFTPClient -> IO ()) -> Expectation
 xftpTest test = runXFTPTest test `shouldReturn` ()

tests/XFTPServerTests.hs

@@ -32,8 +32,9 @@ import Simplex.Messaging.Protocol (BasicAuth, EntityId (..), pattern NoEntity)
 import Simplex.Messaging.Server.Expiration (ExpirationConfig (..))
 import System.Directory (createDirectoryIfMissing, removeDirectoryRecursive, removeFile)
 import System.FilePath ((</>))
-import Test.Hspec
+import Test.Hspec hiding (fit, it)
 import UnliftIO.STM
+import Util
 import XFTPClient
 
 xftpServerTests :: Spec