packages feed

simplexmq 0.3.2 → 0.4.0

raw patch · 31 files changed

+2223/−1070 lines, 31 filesdep +compositiondep +constraintsdep +direct-sqlitedep ~ansi-terminaldep ~cryptonitedep ~generic-random

Dependencies added: composition, constraints, direct-sqlite, file-embed

Dependency ranges changed: ansi-terminal, cryptonite, generic-random, optparse-applicative, random

Files

CHANGELOG.md view
@@ -1,3 +1,27 @@+# 0.4.0++- SMP server implementation and [SMP protocol](https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md) changes:+  - support 3072 bit RSA key size+  - add SMP queue quotas+  - set default transport block size to 4096 bits+  - allow SMP client to change transport block size during transport connection handshake+- SMP agent implementation and protocol changes:+  - additional connection confirmation step for initiating party+  - automatically resume subscribed duplex connections once transport connection is resumed+  - passing an arbitrary binary information between parties during the duplex connection handshake - can be used to identify parties+  - asynchronous duplex connection handshake - the parties do not have to be online at the same time+  - asynchronous message delivery - the agent does not need transport connection to accept client messages for delivery+  - additional confirmation of message reception from the client to prevent message loss in case of process termination+  - set transport block size to 8192 bits (in the future the agent protocol can allow to have different block sizes for different duplex connections)+  - added client commands and notifications (see [agent protocol](https://github.com/simplex-chat/simplexmq/blob/master/protocol/agent-protocol.md)):+    - `REQ` - the notification about joining party establishing connection+    - `ACPT` - the command to accept connection with the joining party+    - `INFO` - the notification with the information from the initiating party+    - `DOWN`/`UP` - the notifications about losing/resuming the connection+    - `ACK` - the command to confirm that the message reception/processing is complete+    - `MID` - the response to `SEND` confirming that the message is accepted by the agent+    - `MERR` - the notification about permanent message delivery error (e.g., `ERR AUTH` indicating that the queue was removed)+ # 0.3.2  - Support websockets
README.md view
@@ -50,11 +50,48 @@  ## Using SMP server and SMP agent -You can either run SMP server locally or try local SMP agent with the deployed demo server:+You can either run your own SMP server locally or deploy using [Linode StackScript](#deploy-smp-server-on-linode), or try local SMP agent with the deployed demo server:  `smp1.simplex.im:5223#pLdiGvm0jD1CMblnov6Edd/391OrYsShw+RgdfR0ChA=`  It's the easiest to try SMP agent via a prototype [simplex-chat](https://github.com/simplex-chat/simplex-chat) terminal UI.++[<img alt="linode" src="./img/linode.svg" align="right" width="200">](https://cloud.linode.com/stackscripts/748014)++## Deploy SMP server on Linode++You can get Linode [free credits](https://www.linode.com/lp/affiliate-referral/?irclickid=02-QkdTEpxyLW0W0EOSREQreUkB2DtzGE2lGTE0&irgwc=1&utm_source=impact) to deploy SMP server.++Deployment on [Linode](https://www.linode.com/) is performed via StackScripts, which serve as recipes for Linode instances, also called Linodes. To deploy SMP server on Linode:++- Create a Linode account or login with an already existing one.+- Open [SMP server StackScript](https://cloud.linode.com/stackscripts/748014) and click "Deploy New Linode".+- You can optionally configure the following parameters:+    - [SMP Server store log](#SMP-server) flag for queue persistence on server restart (recommended).+    - [Linode API token](https://www.linode.com/docs/guides/getting-started-with-the-linode-api#get-an-access-token) for attaching server info as tags to Linode (server address, public key hash, version) and adding A record to your 2nd level domain (Note: 2nd level e.g. `example.com` domain should be [created](https://cloud.linode.com/domains/create) in your account prior to deployment). The API token access scope should be read/write access to "linodes" (to update linode tags - you need them), and "domains" (to add A record for the 3rd level domain, e.g. `smp`).+    - Domain name to use instead of Linode ip address, e.g. `smp.example.com`.+- Choose the region and plan according to your requirements (for regular use Shared CPU Nanode should be sufficient).+- Provide ssh key to be able to connect to your Linode via ssh. This step is required if you haven't provided a Linode API token, because you will need to login to your Linode and get a public key hash either from the welcome message or from the file `/etc/opt/simplex/pub_key_hash` on your Linode after SMP server starts.+- Deploy your Linode. After it starts wait for SMP server to start and for tags to appear (if a Linode API token was provided). It may take up to 5 minutes depending on the connection speed on the Linode. Connecting Linode IP address to provided domain name may take some additional time.+- Get `hostname` and `hash` either from Linode tags (click on a tag and copy it's value from the browser search panel) or via ssh. Linode has a good [guide](https://www.linode.com/docs/guides/use-public-key-authentication-with-ssh/) about ssh.+- Great, your own SMP server is ready! Use `address#hash` as SMP server address in the client.++Please submit an [issue](https://github.com/simplex-chat/simplexmq/issues) if any problems occur.++[<img alt="linode" src="./img/digitalocean.png" align="right" width="300">](https://marketplace.digitalocean.com/apps/simplex-server)++## Deploy SMP server on DigitalOcean++You can deploy SMP server using [SimpleX Server 1-click app](https://marketplace.digitalocean.com/apps/simplex-server) from DigitalOcean marketplace:++- Create a DigitalOcean account or login with an already existing one.+- Click 'Create SimpleX server Droplet' button.+- Choose the region and plan according to your requirements (cheapest Regular plan should be sufficient).+- Provide ssh key and confirm Droplet creation. +- SSH to created Droplet (`ssh root@<droplet_ip_address>`) to get SMP server public key hash - either from the welcome message or from `/etc/opt/simplex/pub_key_hash`. DigitalOcean has a good guide on [how to login to Droplet via ssh](https://docs.digitalocean.com/products/droplets/how-to/connect-with-ssh/).+- Great, your own SMP server is ready! Use `ip_address#hash` as SMP server address in the client.++Please submit an [issue](https://github.com/simplex-chat/simplexmq/issues) if any problems occur.  ## SMP server design
apps/smp-agent/Main.hs view
@@ -8,20 +8,10 @@ import qualified Data.List.NonEmpty as L import Simplex.Messaging.Agent (runSMPAgent) import Simplex.Messaging.Agent.Env.SQLite-import Simplex.Messaging.Client (smpDefaultConfig) import Simplex.Messaging.Transport (TCP, Transport (..))  cfg :: AgentConfig-cfg =-  AgentConfig-    { tcpPort = "5224",-      smpServers = L.fromList ["localhost:5223#KXNE1m2E1m0lm92WGKet9CL6+lO742Vy5G6nsrkvgs8="],-      rsaKeySize = 2048 `div` 8,-      connIdBytes = 12,-      tbqSize = 16,-      dbFile = "smp-agent.db",-      smpCfg = smpDefaultConfig-    }+cfg = defaultAgentConfig {smpServers = L.fromList ["localhost:5223#bU0K+bRg24xWW//lS0umO1Zdw/SXqpJNtm1/RrPLViE="]}  logCfg :: LogConfig logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
apps/smp-server/Main.hs view
@@ -29,19 +29,25 @@ import System.Exit (exitFailure) import System.FilePath (combine) import System.IO (IOMode (..), hFlush, stdout)+import Text.Read (readEither)  defaultServerPort :: ServiceName defaultServerPort = "5223" +defaultBlockSize :: Int+defaultBlockSize = 4096+ serverConfig :: ServerConfig serverConfig =   ServerConfig     { tbqSize = 16,+      msgQueueQuota = 256,       queueIdBytes = 12,       msgIdBytes = 6,       -- below parameters are set based on ini file /etc/opt/simplex/smp-server.ini       transports = undefined,       storeLog = undefined,+      blockSize = undefined,       serverPrivateKey = undefined     } @@ -96,9 +102,9 @@   pure $ makeConfig ini pk storeLog  makeConfig :: IniOpts -> C.FullPrivateKey -> Maybe (StoreLog 'ReadMode) -> ServerConfig-makeConfig IniOpts {serverPort, enableWebsockets} pk storeLog =+makeConfig IniOpts {serverPort, blockSize, enableWebsockets} pk storeLog =   let transports = (serverPort, transport @TCP) : [("80", transport @WS) | enableWebsockets]-   in serverConfig {serverPrivateKey = pk, storeLog, transports}+   in serverConfig {serverPrivateKey = pk, storeLog, blockSize, transports}  printConfig :: ServerConfig -> IO () printConfig ServerConfig {serverPrivateKey, storeLog} = do@@ -109,6 +115,7 @@  initializeServer :: ServerOpts -> IO ServerConfig initializeServer opts = do+  createDirectoryIfMissing False cfgDir   ini <- createIni opts   pk <- createKey ini   storeLog <- openStoreLog opts ini@@ -138,6 +145,7 @@     storeLogFile :: FilePath,     serverKeyFile :: FilePath,     serverPort :: ServiceName,+    blockSize :: Int,     enableWebsockets :: Bool   } @@ -150,7 +158,8 @@       serverKeyFile = opt defaultKeyFile "TRANSPORT" "key_file" ini       serverPort = opt defaultServerPort "TRANSPORT" "port" ini       enableWebsockets = (== Right "on") $ lookupValue "TRANSPORT" "websockets" ini-  pure IniOpts {enableStoreLog, storeLogFile, serverKeyFile, serverPort, enableWebsockets}+  blockSize <- liftEither . readEither $ opt (show defaultBlockSize) "TRANSPORT" "block_size" ini+  pure IniOpts {enableStoreLog, storeLogFile, serverKeyFile, serverPort, blockSize, enableWebsockets}   where     opt :: String -> Text -> Text -> Ini -> String     opt def section key ini = either (const def) T.unpack $ lookupValue section key ini@@ -177,6 +186,9 @@          \# port: "       <> defaultServerPort       <> "\n\+         \# block_size: "+      <> show defaultBlockSize+      <> "\n\          \websockets: on\n"   pure     IniOpts@@ -184,6 +196,7 @@         storeLogFile = defaultStoreLogFile,         serverKeyFile = defaultKeyFile,         serverPort = defaultServerPort,+        blockSize = defaultBlockSize,         enableWebsockets = True       } @@ -201,7 +214,6 @@  createKey :: IniOpts -> IO C.FullPrivateKey createKey IniOpts {serverKeyFile} = do-  createDirectoryIfMissing True cfgDir   (_, pk) <- C.generateKeyPair newKeySize   S.writeKeyFile S.TraditionalFormat serverKeyFile [PrivKeyRSA $ C.rsaPrivateKey pk]   pure pk@@ -222,7 +234,7 @@   when (map toLower ok /= "y") exitFailure  serverKeyHash :: C.FullPrivateKey -> B.ByteString-serverKeyHash = encode . C.unKeyHash . C.publicKeyHash . C.publicKey+serverKeyHash = encode . C.unKeyHash . C.publicKeyHash . C.publicKey'  openStoreLog :: ServerOpts -> IniOpts -> IO (Maybe (StoreLog 'ReadMode)) openStoreLog ServerOpts {enableStoreLog = l} IniOpts {enableStoreLog = l', storeLogFile = f}
simplexmq.cabal view
@@ -1,13 +1,13 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.33.0.+-- This file has been generated from package.yaml by hpack version 0.34.4. -- -- see: https://github.com/sol/hpack ----- hash: 1f213808753ff701ca8bcdc7b3296d99346fdbe9e0b6fbcfb372f928bcffa525+-- hash: e25d53dc1b12c8bcdf029091182613198be5271348ea19174b6532d820fe1fc9  name:           simplexmq-version:        0.3.2+version:        0.4.0 synopsis:       SimpleXMQ message broker description:    This package includes <./docs/Simplex-Messaging-Server.html server>,                 <./docs/Simplex-Messaging-Client.html client> and@@ -35,9 +35,10 @@       Simplex.Messaging.Agent.Client       Simplex.Messaging.Agent.Env.SQLite       Simplex.Messaging.Agent.Protocol+      Simplex.Messaging.Agent.RetryInterval       Simplex.Messaging.Agent.Store       Simplex.Messaging.Agent.Store.SQLite-      Simplex.Messaging.Agent.Store.SQLite.Schema+      Simplex.Messaging.Agent.Store.SQLite.Migrations       Simplex.Messaging.Client       Simplex.Messaging.Crypto       Simplex.Messaging.Parsers@@ -59,7 +60,7 @@   ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns   build-depends:       QuickCheck ==2.14.*-    , ansi-terminal ==0.10.*+    , ansi-terminal >=0.10 && <0.12     , asn1-encoding ==0.9.*     , asn1-types ==0.3.*     , async ==2.2.*@@ -67,17 +68,21 @@     , base >=4.7 && <5     , base64-bytestring >=1.0 && <1.3     , bytestring ==0.10.*+    , composition ==1.0.*+    , constraints >=0.12 && <0.14     , containers ==0.6.*-    , cryptonite ==0.27.*+    , cryptonite >=0.27 && <0.30+    , direct-sqlite ==2.3.*     , directory ==1.3.*+    , file-embed ==0.0.14.*     , filepath ==1.4.*-    , generic-random ==1.3.*+    , generic-random >=1.3 && <1.5     , iso8601-time ==0.1.*     , memory ==0.15.*     , mtl ==2.2.*     , network ==3.1.*     , network-transport ==0.5.*-    , random ==1.1.*+    , random >=1.1 && <1.3     , simple-logger ==0.1.*     , sqlite-simple ==0.4.*     , stm ==2.5.*@@ -100,7 +105,7 @@   ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded   build-depends:       QuickCheck ==2.14.*-    , ansi-terminal ==0.10.*+    , ansi-terminal >=0.10 && <0.12     , asn1-encoding ==0.9.*     , asn1-types ==0.3.*     , async ==2.2.*@@ -108,17 +113,21 @@     , base >=4.7 && <5     , base64-bytestring >=1.0 && <1.3     , bytestring ==0.10.*+    , composition ==1.0.*+    , constraints >=0.12 && <0.14     , containers ==0.6.*-    , cryptonite ==0.27.*+    , cryptonite >=0.27 && <0.30+    , direct-sqlite ==2.3.*     , directory ==1.3.*+    , file-embed ==0.0.14.*     , filepath ==1.4.*-    , generic-random ==1.3.*+    , generic-random >=1.3 && <1.5     , iso8601-time ==0.1.*     , memory ==0.15.*     , mtl ==2.2.*     , network ==3.1.*     , network-transport ==0.5.*-    , random ==1.1.*+    , random >=1.1 && <1.3     , simple-logger ==0.1.*     , simplexmq     , sqlite-simple ==0.4.*@@ -142,7 +151,7 @@   ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded   build-depends:       QuickCheck ==2.14.*-    , ansi-terminal ==0.10.*+    , ansi-terminal >=0.10 && <0.12     , asn1-encoding ==0.9.*     , asn1-types ==0.3.*     , async ==2.2.*@@ -150,20 +159,24 @@     , base >=4.7 && <5     , base64-bytestring >=1.0 && <1.3     , bytestring ==0.10.*+    , composition ==1.0.*+    , constraints >=0.12 && <0.14     , containers ==0.6.*-    , cryptonite ==0.27.*+    , cryptonite >=0.27 && <0.30     , cryptostore ==0.2.*+    , direct-sqlite ==2.3.*     , directory ==1.3.*+    , file-embed ==0.0.14.*     , filepath ==1.4.*-    , generic-random ==1.3.*+    , generic-random >=1.3 && <1.5     , ini ==0.4.*     , iso8601-time ==0.1.*     , memory ==0.15.*     , mtl ==2.2.*     , network ==3.1.*     , network-transport ==0.5.*-    , optparse-applicative ==0.15.*-    , random ==1.1.*+    , optparse-applicative >=0.15 && <0.17+    , random >=1.1 && <1.3     , simple-logger ==0.1.*     , simplexmq     , sqlite-simple ==0.4.*@@ -183,6 +196,7 @@   main-is: Test.hs   other-modules:       AgentTests+      AgentTests.FunctionalAPITests       AgentTests.SQLiteTests       ProtocolErrorTests       ServerTests@@ -195,7 +209,7 @@   build-depends:       HUnit ==1.6.*     , QuickCheck ==2.14.*-    , ansi-terminal ==0.10.*+    , ansi-terminal >=0.10 && <0.12     , asn1-encoding ==0.9.*     , asn1-types ==0.3.*     , async ==2.2.*@@ -203,11 +217,15 @@     , base >=4.7 && <5     , base64-bytestring >=1.0 && <1.3     , bytestring ==0.10.*+    , composition ==1.0.*+    , constraints >=0.12 && <0.14     , containers ==0.6.*-    , cryptonite ==0.27.*+    , cryptonite >=0.27 && <0.30+    , direct-sqlite ==2.3.*     , directory ==1.3.*+    , file-embed ==0.0.14.*     , filepath ==1.4.*-    , generic-random ==1.3.*+    , generic-random >=1.3 && <1.5     , hspec ==2.7.*     , hspec-core ==2.7.*     , iso8601-time ==0.1.*@@ -215,7 +233,7 @@     , mtl ==2.2.*     , network ==3.1.*     , network-transport ==0.5.*-    , random ==1.1.*+    , random >=1.1 && <1.3     , simple-logger ==0.1.*     , simplexmq     , sqlite-simple ==0.4.*
src/Simplex/Messaging/Agent.hs view
@@ -1,12 +1,17 @@+{-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE GADTs #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE NumericUnderscores #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-}+{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TupleSections #-}  -- | -- Module      : Simplex.Messaging.Agent@@ -21,10 +26,29 @@ -- -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/agent-protocol.md module Simplex.Messaging.Agent-  ( runSMPAgent,+  ( -- * SMP agent over TCP+    runSMPAgent,     runSMPAgentBlocking,++    -- * queue-based SMP agent+    getAgentClient,+    runAgentClient,++    -- * SMP agent functional API+    AgentClient (..),+    AgentMonad,+    AgentErrorMonad,     getSMPAgentClient,-    runSMPAgentClient,+    disconnectAgentClient, -- used in tests+    withAgentLock,+    createConnection,+    joinConnection,+    acceptConnection,+    subscribeConnection,+    sendMessage,+    ackMessage,+    suspendConnection,+    deleteConnection,   ) where @@ -34,10 +58,16 @@ import Control.Monad.IO.Unlift (MonadUnliftIO) import Control.Monad.Reader import Crypto.Random (MonadRandom)+import Data.Bifunctor (second) import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B+import Data.Composition ((.:), (.:.))+import Data.Functor (($>)) import Data.List.NonEmpty (NonEmpty (..)) import qualified Data.List.NonEmpty as L+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as M+import Data.Maybe (isJust) import qualified Data.Text as T import Data.Text.Encoding (decodeUtf8) import Data.Time.Clock@@ -45,16 +75,17 @@ import Simplex.Messaging.Agent.Client import Simplex.Messaging.Agent.Env.SQLite import Simplex.Messaging.Agent.Protocol+import Simplex.Messaging.Agent.RetryInterval import Simplex.Messaging.Agent.Store-import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore, connectSQLiteStore)+import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore) import Simplex.Messaging.Client (SMPServerTransmission) import qualified Simplex.Messaging.Crypto as C-import Simplex.Messaging.Protocol (CorrId (..), MsgBody, SenderPublicKey)+import Simplex.Messaging.Protocol (MsgBody, SenderPublicKey) import qualified Simplex.Messaging.Protocol as SMP import Simplex.Messaging.Transport (ATransport (..), TProxy, Transport (..), runTransportServer)-import Simplex.Messaging.Util (bshow)+import Simplex.Messaging.Util (bshow, tryError) import System.Random (randomR)-import UnliftIO.Async (race_)+import UnliftIO.Async (Async, async, race_) import qualified UnliftIO.Exception as E import UnliftIO.STM @@ -75,19 +106,68 @@   where     smpAgent :: forall c m'. (Transport c, MonadUnliftIO m', MonadReader Env m') => TProxy c -> m' ()     smpAgent _ = runTransportServer started tcpPort $ \(h :: c) -> do-      liftIO $ putLn h "Welcome to SMP v0.3.2 agent"-      c <- getSMPAgentClient+      liftIO $ putLn h "Welcome to SMP v0.4.0 agent"+      c <- getAgentClient       logConnection c True-      race_ (connectClient h c) (runSMPAgentClient c)-        `E.finally` (closeSMPServerClients c >> logConnection c False)+      race_ (connectClient h c) (runAgentClient c)+        `E.finally` disconnectAgentClient c --- | Creates an SMP agent instance that receives commands and sends responses via 'TBQueue's.-getSMPAgentClient :: (MonadUnliftIO m, MonadReader Env m) => m AgentClient-getSMPAgentClient = do-  n <- asks clientCounter-  cfg <- asks config-  atomically $ newAgentClient n cfg+-- | Creates an SMP agent client instance+getSMPAgentClient :: (MonadRandom m, MonadUnliftIO m) => AgentConfig -> m AgentClient+getSMPAgentClient cfg = newSMPAgentEnv cfg >>= runReaderT runAgent+  where+    runAgent = do+      c <- getAgentClient+      action <- async $ subscriber c `E.finally` disconnectAgentClient c+      pure c {smpSubscriber = action} +disconnectAgentClient :: MonadUnliftIO m => AgentClient -> m ()+disconnectAgentClient c = closeAgentClient c >> logConnection c False++-- |+type AgentErrorMonad m = (MonadUnliftIO m, MonadError AgentErrorType m)++-- | Create SMP agent connection (NEW command)+createConnection :: AgentErrorMonad m => AgentClient -> m (ConnId, SMPQueueInfo)+createConnection c = withAgentEnv c $ newConn c ""++-- | Join SMP agent connection (JOIN command)+joinConnection :: AgentErrorMonad m => AgentClient -> SMPQueueInfo -> ConnInfo -> m ConnId+joinConnection c = withAgentEnv c .: joinConn c ""++-- | Approve confirmation (LET command)+acceptConnection :: AgentErrorMonad m => AgentClient -> ConnId -> ConfirmationId -> ConnInfo -> m ()+acceptConnection c = withAgentEnv c .:. acceptConnection' c++-- | Subscribe to receive connection messages (SUB command)+subscribeConnection :: AgentErrorMonad m => AgentClient -> ConnId -> m ()+subscribeConnection c = withAgentEnv c . subscribeConnection' c++-- | Send message to the connection (SEND command)+sendMessage :: AgentErrorMonad m => AgentClient -> ConnId -> MsgBody -> m AgentMsgId+sendMessage c = withAgentEnv c .: sendMessage' c++ackMessage :: AgentErrorMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()+ackMessage c = withAgentEnv c .: ackMessage' c++-- | Suspend SMP agent connection (OFF command)+suspendConnection :: AgentErrorMonad m => AgentClient -> ConnId -> m ()+suspendConnection c = withAgentEnv c . suspendConnection' c++-- | Delete SMP agent connection (DEL command)+deleteConnection :: AgentErrorMonad m => AgentClient -> ConnId -> m ()+deleteConnection c = withAgentEnv c . deleteConnection' c++withAgentEnv :: AgentClient -> ReaderT Env m a -> m a+withAgentEnv c = (`runReaderT` agentEnv c)++-- withAgentClient :: AgentErrorMonad m => AgentClient -> ReaderT Env m a -> m a+-- withAgentClient c = withAgentLock c . withAgentEnv c++-- | Creates an SMP agent client instance that receives commands and sends responses via 'TBQueue's.+getAgentClient :: (MonadUnliftIO m, MonadReader Env m) => m AgentClient+getAgentClient = ask >>= atomically . newAgentClient+ connectClient :: Transport c => MonadUnliftIO m => c -> AgentClient -> m () connectClient h c = race_ (send h c) (receive h c) @@ -97,19 +177,15 @@    in logInfo $ T.unwords ["client", showText (clientId c), event, "Agent"]  -- | Runs an SMP agent instance that receives commands and sends responses via 'TBQueue's.-runSMPAgentClient :: (MonadUnliftIO m, MonadReader Env m) => AgentClient -> m ()-runSMPAgentClient c = do-  db <- asks $ dbFile . config-  s1 <- connectSQLiteStore db-  s2 <- connectSQLiteStore db-  race_ (subscriber c s1) (client c s2)+runAgentClient :: (MonadUnliftIO m, MonadReader Env m) => AgentClient -> m ()+runAgentClient c = race_ (subscriber c) (client c)  receive :: forall c m. (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()-receive h c@AgentClient {rcvQ, sndQ} = forever $ do-  (corrId, cAlias, cmdOrErr) <- tGet SClient h+receive h c@AgentClient {rcvQ, subQ} = forever $ do+  (corrId, connId, cmdOrErr) <- tGet SClient h   case cmdOrErr of-    Right cmd -> write rcvQ (corrId, cAlias, cmd)-    Left e -> write sndQ (corrId, cAlias, ERR e)+    Right cmd -> write rcvQ (corrId, connId, cmd)+    Left e -> write subQ (corrId, connId, ERR e)   where     write :: TBQueue (ATransmission p) -> ATransmission p -> m ()     write q t = do@@ -117,193 +193,330 @@       atomically $ writeTBQueue q t  send :: (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()-send h c@AgentClient {sndQ} = forever $ do-  t <- atomically $ readTBQueue sndQ+send h c@AgentClient {subQ} = forever $ do+  t <- atomically $ readTBQueue subQ   tPut h t   logClient c "<--" t  logClient :: MonadUnliftIO m => AgentClient -> ByteString -> ATransmission a -> m ()-logClient AgentClient {clientId} dir (CorrId corrId, cAlias, cmd) = do-  logInfo . decodeUtf8 $ B.unwords [bshow clientId, dir, "A :", corrId, cAlias, B.takeWhile (/= ' ') $ serializeCommand cmd]+logClient AgentClient {clientId} dir (corrId, connId, cmd) = do+  logInfo . decodeUtf8 $ B.unwords [bshow clientId, dir, "A :", corrId, connId, B.takeWhile (/= ' ') $ serializeCommand cmd] -client :: (MonadUnliftIO m, MonadReader Env m) => AgentClient -> SQLiteStore -> m ()-client c@AgentClient {rcvQ, sndQ} st = forever $ do-  t@(corrId, cAlias, _) <- atomically $ readTBQueue rcvQ-  runExceptT (processCommand c st t) >>= \case-    Left e -> atomically $ writeTBQueue sndQ (corrId, cAlias, ERR e)-    Right _ -> return ()+client :: forall m. (MonadUnliftIO m, MonadReader Env m) => AgentClient -> m ()+client c@AgentClient {rcvQ, subQ} = forever $ do+  (corrId, connId, cmd) <- atomically $ readTBQueue rcvQ+  withAgentLock c (runExceptT $ processCommand c (connId, cmd))+    >>= atomically . writeTBQueue subQ . \case+      Left e -> (corrId, connId, ERR e)+      Right (connId', resp) -> (corrId, connId', resp)  withStore ::   AgentMonad m =>-  (forall m'. (MonadUnliftIO m', MonadError StoreError m') => m' a) ->+  (forall m'. (MonadUnliftIO m', MonadError StoreError m') => SQLiteStore -> m' a) ->   m a withStore action = do-  runExceptT (action `E.catch` handleInternal) >>= \case+  st <- asks store+  runExceptT (action st `E.catch` handleInternal) >>= \case     Right c -> return c     Left e -> throwError $ storeError e   where+    -- TODO when parsing exception happens in store, the agent hangs;+    -- changing SQLError to SomeException does not help     handleInternal :: (MonadError StoreError m') => SQLError -> m' a     handleInternal e = throwError . SEInternal $ bshow e     storeError :: StoreError -> AgentErrorType     storeError = \case-      SEConnNotFound -> CONN UNKNOWN+      SEConnNotFound -> CONN NOT_FOUND       SEConnDuplicate -> CONN DUPLICATE+      SEBadConnType CRcv -> CONN SIMPLEX+      SEBadConnType CSnd -> CONN SIMPLEX       e -> INTERNAL $ show e -processCommand :: forall m. AgentMonad m => AgentClient -> SQLiteStore -> ATransmission 'Client -> m ()-processCommand c@AgentClient {sndQ} st (corrId, connAlias, cmd) =-  case cmd of-    NEW -> createNewConnection-    JOIN smpQueueInfo replyMode -> joinConnection smpQueueInfo replyMode-    SUB -> subscribeConnection connAlias-    SUBALL -> subscribeAll-    SEND msgBody -> sendMessage msgBody-    OFF -> suspendConnection-    DEL -> deleteConnection+-- | execute any SMP agent command+processCommand :: forall m. AgentMonad m => AgentClient -> (ConnId, ACommand 'Client) -> m (ConnId, ACommand 'Agent)+processCommand c (connId, cmd) = case cmd of+  NEW -> second INV <$> newConn c connId+  JOIN smpQueueInfo connInfo -> (,OK) <$> joinConn c connId smpQueueInfo connInfo+  ACPT confId ownConnInfo -> acceptConnection' c connId confId ownConnInfo $> (connId, OK)+  SUB -> subscribeConnection' c connId $> (connId, OK)+  SEND msgBody -> (connId,) . MID <$> sendMessage' c connId msgBody+  ACK msgId -> ackMessage' c connId msgId $> (connId, OK)+  OFF -> suspendConnection' c connId $> (connId, OK)+  DEL -> deleteConnection' c connId $> (connId, OK)++newConn :: AgentMonad m => AgentClient -> ConnId -> m (ConnId, SMPQueueInfo)+newConn c connId = do+  srv <- getSMPServer+  (rq, qInfo) <- newRcvQueue c srv+  g <- asks idsDrg+  let cData = ConnData {connId}+  connId' <- withStore $ \st -> createRcvConn st g cData rq+  addSubscription c rq connId'+  pure (connId', qInfo)++joinConn :: AgentMonad m => AgentClient -> ConnId -> SMPQueueInfo -> ConnInfo -> m ConnId+joinConn c connId qInfo cInfo = do+  (sq, senderKey, verifyKey) <- newSndQueue qInfo+  g <- asks idsDrg+  cfg <- asks config+  let cData = ConnData {connId}+  connId' <- withStore $ \st -> createSndConn st g cData sq+  confirmQueue c sq senderKey cInfo+  activateQueueJoining c connId' sq verifyKey $ retryInterval cfg+  pure connId'++activateQueueJoining :: forall m. AgentMonad m => AgentClient -> ConnId -> SndQueue -> VerificationKey -> RetryInterval -> m ()+activateQueueJoining c connId sq verifyKey retryInterval =+  activateQueue c connId sq verifyKey retryInterval createReplyQueue   where-    createNewConnection :: m ()-    createNewConnection = do-      -- TODO create connection alias if not passed-      -- make connAlias Maybe?+    createReplyQueue :: m ()+    createReplyQueue = do       srv <- getSMPServer-      (rq, qInfo) <- newReceiveQueue c srv connAlias-      withStore $ createRcvConn st rq-      respond $ INV qInfo--    getSMPServer :: m SMPServer-    getSMPServer =-      asks (smpServers . config) >>= \case-        srv :| [] -> pure srv-        servers -> do-          gen <- asks randomServer-          i <- atomically . stateTVar gen $ randomR (0, L.length servers - 1)-          pure $ servers L.!! i+      (rq, qInfo') <- newRcvQueue c srv+      addSubscription c rq connId+      withStore $ \st -> upgradeSndConnToDuplex st connId rq+      sendControlMessage c sq $ REPLY qInfo' -    joinConnection :: SMPQueueInfo -> ReplyMode -> m ()-    joinConnection qInfo (ReplyMode replyMode) = do-      -- TODO create connection alias if not passed-      -- make connAlias Maybe?-      (sq, senderKey, verifyKey) <- newSendQueue qInfo connAlias-      withStore $ createSndConn st sq-      connectToSendQueue c st sq senderKey verifyKey-      when (replyMode == On) $ createReplyQueue sq-    -- TODO this response is disabled to avoid two responses in terminal client (OK + CON),-    -- respond OK+-- | Approve confirmation (LET command) in Reader monad+acceptConnection' :: AgentMonad m => AgentClient -> ConnId -> ConfirmationId -> ConnInfo -> m ()+acceptConnection' c connId confId ownConnInfo =+  withStore (`getConn` connId) >>= \case+    SomeConn SCRcv (RcvConnection _ rq) -> do+      AcceptedConfirmation {senderKey} <- withStore $ \st -> acceptConfirmation st confId ownConnInfo+      processConfirmation c rq senderKey+    _ -> throwError $ CMD PROHIBITED -    subscribeConnection :: ConnAlias -> m ()-    subscribeConnection cAlias =-      withStore (getConn st cAlias) >>= \case-        SomeConn _ (DuplexConnection _ rq _) -> subscribe rq-        SomeConn _ (RcvConnection _ rq) -> subscribe rq-        _ -> throwError $ CONN SIMPLEX-      where-        subscribe rq = subscribeQueue c rq cAlias >> respond' cAlias OK+processConfirmation :: AgentMonad m => AgentClient -> RcvQueue -> SenderPublicKey -> m ()+processConfirmation c rq sndKey = do+  withStore $ \st -> setRcvQueueStatus st rq Confirmed+  secureQueue c rq sndKey+  withStore $ \st -> setRcvQueueStatus st rq Secured -    -- TODO remove - hack for subscribing to all; respond' and parameterization of subscribeConnection are byproduct-    subscribeAll :: m ()-    subscribeAll = withStore (getAllConnAliases st) >>= mapM_ subscribeConnection+-- | Subscribe to receive connection messages (SUB command) in Reader monad+subscribeConnection' :: forall m. AgentMonad m => AgentClient -> ConnId -> m ()+subscribeConnection' c connId =+  withStore (`getConn` connId) >>= \case+    SomeConn _ (DuplexConnection _ rq sq) -> do+      resumeDelivery sq+      case status (sq :: SndQueue) of+        Confirmed -> withVerifyKey sq $ \verifyKey -> do+          conf <- withStore (`getAcceptedConfirmation` connId)+          secureQueue c rq $ senderKey (conf :: AcceptedConfirmation)+          withStore $ \st -> setRcvQueueStatus st rq Secured+          activateSecuredQueue rq sq verifyKey+        Secured -> withVerifyKey sq $ activateSecuredQueue rq sq+        Active -> subscribeQueue c rq connId+        _ -> throwError $ INTERNAL "unexpected queue status"+    SomeConn _ (SndConnection _ sq) -> do+      resumeDelivery sq+      case status (sq :: SndQueue) of+        Confirmed -> withVerifyKey sq $ \verifyKey ->+          activateQueueJoining c connId sq verifyKey =<< resumeInterval+        Active -> throwError $ CONN SIMPLEX+        _ -> throwError $ INTERNAL "unexpected queue status"+    SomeConn _ (RcvConnection _ rq) -> subscribeQueue c rq connId+  where+    resumeDelivery :: SndQueue -> m ()+    resumeDelivery SndQueue {server} = do+      wasDelivering <- resumeMsgDelivery c connId server+      unless wasDelivering $ do+        pending <- withStore (`getPendingMsgs` connId)+        queuePendingMsgs c connId pending+    withVerifyKey :: SndQueue -> (C.PublicKey -> m ()) -> m ()+    withVerifyKey sq action =+      let err = throwError $ INTERNAL "missing signing key public counterpart"+       in maybe err action . C.publicKey $ signKey sq+    activateSecuredQueue :: RcvQueue -> SndQueue -> C.PublicKey -> m ()+    activateSecuredQueue rq sq verifyKey = do+      activateQueueInitiating c connId sq verifyKey =<< resumeInterval+      subscribeQueue c rq connId+    resumeInterval :: m RetryInterval+    resumeInterval = do+      r <- asks $ retryInterval . config+      pure r {initialInterval = 5_000_000} -    sendMessage :: MsgBody -> m ()-    sendMessage msgBody =-      withStore (getConn st connAlias) >>= \case-        SomeConn _ (DuplexConnection _ _ sq) -> sendMsg sq-        SomeConn _ (SndConnection _ sq) -> sendMsg sq-        _ -> throwError $ CONN SIMPLEX+-- | Send message to the connection (SEND command) in Reader monad+sendMessage' :: forall m. AgentMonad m => AgentClient -> ConnId -> MsgBody -> m AgentMsgId+sendMessage' c connId msg =+  withStore (`getConn` connId) >>= \case+    SomeConn _ (DuplexConnection _ _ sq) -> enqueueMessage sq+    SomeConn _ (SndConnection _ sq) -> enqueueMessage sq+    _ -> throwError $ CONN SIMPLEX+  where+    enqueueMessage :: SndQueue -> m AgentMsgId+    enqueueMessage SndQueue {server} = do+      msgId <- storeSentMsg+      wasDelivering <- resumeMsgDelivery c connId server+      pending <-+        if wasDelivering+          then pure [PendingMsg {connId, msgId}]+          else withStore (`getPendingMsgs` connId)+      queuePendingMsgs c connId pending+      pure $ unId msgId       where-        sendMsg sq = do+        storeSentMsg :: m InternalId+        storeSentMsg = do           internalTs <- liftIO getCurrentTime-          (internalId, internalSndId, previousMsgHash) <- withStore $ updateSndIds st sq-          let msgStr =-                serializeSMPMessage-                  SMPMessage-                    { senderMsgId = unSndId internalSndId,-                      senderTimestamp = internalTs,-                      previousMsgHash,-                      agentMessage = A_MSG msgBody-                    }-              msgHash = C.sha256Hash msgStr-          withStore $-            createSndMsg st sq $-              SndMsgData {internalId, internalSndId, internalTs, msgBody, internalHash = msgHash}-          sendAgentMessage c sq msgStr-          respond $ SENT (unId internalId)+          withStore $ \st -> do+            (internalId, internalSndId, previousMsgHash) <- updateSndIds st connId+            let msgBody =+                  serializeSMPMessage+                    SMPMessage+                      { senderMsgId = unSndId internalSndId,+                        senderTimestamp = internalTs,+                        previousMsgHash,+                        agentMessage = A_MSG msg+                      }+                internalHash = C.sha256Hash msgBody+                msgData = SndMsgData {..}+            createSndMsg st connId msgData+            pure internalId -    suspendConnection :: m ()-    suspendConnection =-      withStore (getConn st connAlias) >>= \case-        SomeConn _ (DuplexConnection _ rq _) -> suspend rq-        SomeConn _ (RcvConnection _ rq) -> suspend rq-        _ -> throwError $ CONN SIMPLEX-      where-        suspend rq = suspendQueue c rq >> respond OK+resumeMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnId -> SMPServer -> m Bool+resumeMsgDelivery c connId srv = do+  void $ resume srv (srvMsgDeliveries c) $ runSrvMsgDelivery c srv+  resume connId (connMsgDeliveries c) $ runMsgDelivery c connId srv+  where+    resume :: Ord a => a -> TVar (Map a (Async ())) -> m () -> m Bool+    resume key actionMap actionProcess = do+      isDelivering <- isJust . M.lookup key <$> readTVarIO actionMap+      unless isDelivering $+        async actionProcess+          >>= atomically . modifyTVar actionMap . M.insert key+      pure isDelivering -    deleteConnection :: m ()-    deleteConnection =-      withStore (getConn st connAlias) >>= \case-        SomeConn _ (DuplexConnection _ rq _) -> delete rq-        SomeConn _ (RcvConnection _ rq) -> delete rq-        _ -> delConn-      where-        delConn = withStore (deleteConn st connAlias) >> respond OK-        delete rq = do-          deleteQueue c rq-          removeSubscription c connAlias-          delConn+queuePendingMsgs :: AgentMonad m => AgentClient -> ConnId -> [PendingMsg] -> m ()+queuePendingMsgs c connId pending =+  atomically $ getPendingMsgQ connId (connMsgQueues c) >>= forM_ pending . writeTQueue -    createReplyQueue :: SndQueue -> m ()-    createReplyQueue sq = do-      srv <- getSMPServer-      (rq, qInfo) <- newReceiveQueue c srv connAlias-      withStore $ upgradeSndConnToDuplex st connAlias rq-      senderTimestamp <- liftIO getCurrentTime-      sendAgentMessage c sq . serializeSMPMessage $-        SMPMessage-          { senderMsgId = 0,-            senderTimestamp,-            previousMsgHash = "",-            agentMessage = REPLY qInfo-          }+getPendingMsgQ :: Ord a => a -> TVar (Map a (TQueue PendingMsg)) -> STM (TQueue PendingMsg)+getPendingMsgQ key queueMap = do+  maybe newMsgQueue pure . M.lookup key =<< readTVar queueMap+  where+    newMsgQueue :: STM (TQueue PendingMsg)+    newMsgQueue = do+      mq <- newTQueue+      modifyTVar queueMap $ M.insert key mq+      pure mq -    respond :: ACommand 'Agent -> m ()-    respond = respond' connAlias+runMsgDelivery :: AgentMonad m => AgentClient -> ConnId -> SMPServer -> m ()+runMsgDelivery c connId srv = do+  mq <- atomically . getPendingMsgQ connId $ connMsgQueues c+  smq <- atomically . getPendingMsgQ srv $ srvMsgQueues c+  forever . atomically $ readTQueue mq >>= writeTQueue smq -    respond' :: ConnAlias -> ACommand 'Agent -> m ()-    respond' cAlias resp = atomically $ writeTBQueue sndQ (corrId, cAlias, resp)+runSrvMsgDelivery :: forall m. AgentMonad m => AgentClient -> SMPServer -> m ()+runSrvMsgDelivery c@AgentClient {subQ} srv = do+  mq <- atomically . getPendingMsgQ srv $ srvMsgQueues c+  ri <- asks $ reconnectInterval . config+  forever $ do+    PendingMsg {connId, msgId} <- atomically $ readTQueue mq+    let mId = unId msgId+    withStore (\st -> E.try $ getPendingMsgData st connId msgId) >>= \case+      Left (e :: E.SomeException) ->+        notify connId $ MERR mId (INTERNAL $ show e)+      Right (sq, msgBody) -> do+        withRetryInterval ri $ \loop -> do+          tryError (sendAgentMessage c sq msgBody) >>= \case+            Left e -> case e of+              SMP SMP.QUOTA -> loop+              SMP {} -> notify connId $ MERR mId e+              CMD {} -> notify connId $ MERR mId e+              _ -> loop+            Right () -> do+              notify connId $ SENT mId+              withStore $ \st -> updateSndMsgStatus st connId msgId SndMsgSent+  where+    notify :: ConnId -> ACommand 'Agent -> m ()+    notify connId cmd = atomically $ writeTBQueue subQ ("", connId, cmd) -subscriber :: (MonadUnliftIO m, MonadReader Env m) => AgentClient -> SQLiteStore -> m ()-subscriber c@AgentClient {msgQ} st = forever $ do-  -- TODO this will only process messages and notifications+ackMessage' :: forall m. AgentMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()+ackMessage' c connId msgId = do+  withStore (`getConn` connId) >>= \case+    SomeConn _ (DuplexConnection _ rq _) -> ack rq+    SomeConn _ (RcvConnection _ rq) -> ack rq+    _ -> throwError $ CONN SIMPLEX+  where+    ack :: RcvQueue -> m ()+    ack rq = do+      let mId = InternalId msgId+      withStore $ \st -> checkRcvMsg st connId mId+      sendAck c rq+      withStore $ \st -> updateRcvMsgAck st connId mId++-- | Suspend SMP agent connection (OFF command) in Reader monad+suspendConnection' :: AgentMonad m => AgentClient -> ConnId -> m ()+suspendConnection' c connId =+  withStore (`getConn` connId) >>= \case+    SomeConn _ (DuplexConnection _ rq _) -> suspendQueue c rq+    SomeConn _ (RcvConnection _ rq) -> suspendQueue c rq+    _ -> throwError $ CONN SIMPLEX++-- | Delete SMP agent connection (DEL command) in Reader monad+deleteConnection' :: forall m. AgentMonad m => AgentClient -> ConnId -> m ()+deleteConnection' c connId =+  withStore (`getConn` connId) >>= \case+    SomeConn _ (DuplexConnection _ rq _) -> delete rq+    SomeConn _ (RcvConnection _ rq) -> delete rq+    _ -> withStore (`deleteConn` connId)+  where+    delete :: RcvQueue -> m ()+    delete rq = do+      deleteQueue c rq+      removeSubscription c connId+      withStore (`deleteConn` connId)++getSMPServer :: AgentMonad m => m SMPServer+getSMPServer =+  asks (smpServers . config) >>= \case+    srv :| [] -> pure srv+    servers -> do+      gen <- asks randomServer+      i <- atomically . stateTVar gen $ randomR (0, L.length servers - 1)+      pure $ servers L.!! i++sendControlMessage :: AgentMonad m => AgentClient -> SndQueue -> AMessage -> m ()+sendControlMessage c sq agentMessage = do+  senderTimestamp <- liftIO getCurrentTime+  sendAgentMessage c sq . serializeSMPMessage $+    SMPMessage+      { senderMsgId = 0,+        senderTimestamp,+        previousMsgHash = "",+        agentMessage+      }++subscriber :: (MonadUnliftIO m, MonadReader Env m) => AgentClient -> m ()+subscriber c@AgentClient {msgQ} = forever $ do   t <- atomically $ readTBQueue msgQ-  runExceptT (processSMPTransmission c st t) >>= \case+  withAgentLock c (runExceptT $ processSMPTransmission c t) >>= \case     Left e -> liftIO $ print e     Right _ -> return () -processSMPTransmission :: forall m. AgentMonad m => AgentClient -> SQLiteStore -> SMPServerTransmission -> m ()-processSMPTransmission c@AgentClient {sndQ} st (srv, rId, cmd) = do-  withStore (getRcvConn st srv rId) >>= \case-    SomeConn SCDuplex (DuplexConnection _ rq _) -> processSMP SCDuplex rq-    SomeConn SCRcv (RcvConnection _ rq) -> processSMP SCRcv rq-    _ -> atomically $ writeTBQueue sndQ ("", "", ERR $ CONN SIMPLEX)+processSMPTransmission :: forall m. AgentMonad m => AgentClient -> SMPServerTransmission -> m ()+processSMPTransmission c@AgentClient {subQ} (srv, rId, cmd) = do+  withStore (\st -> getRcvConn st srv rId) >>= \case+    SomeConn SCDuplex (DuplexConnection cData rq _) -> processSMP SCDuplex cData rq+    SomeConn SCRcv (RcvConnection cData rq) -> processSMP SCRcv cData rq+    _ -> atomically $ writeTBQueue subQ ("", "", ERR $ CONN NOT_FOUND)   where-    processSMP :: SConnType c -> RcvQueue -> m ()-    processSMP cType rq@RcvQueue {connAlias, status} =+    processSMP :: SConnType c -> ConnData -> RcvQueue -> m ()+    processSMP cType ConnData {connId} rq@RcvQueue {status} =       case cmd of         SMP.MSG srvMsgId srvTs msgBody -> do           -- TODO deduplicate with previously received           msg <- decryptAndVerify rq msgBody           let msgHash = C.sha256Hash msg-          agentMsg <- liftEither $ parseSMPMessage msg-          case agentMsg of-            SMPConfirmation senderKey -> smpConfirmation senderKey-            SMPMessage {agentMessage, senderMsgId, senderTimestamp, previousMsgHash} ->+          case parseSMPMessage msg of+            Left e -> notify $ ERR e+            Right (SMPConfirmation senderKey cInfo) -> smpConfirmation senderKey cInfo >> sendAck c rq+            Right SMPMessage {agentMessage, senderMsgId, senderTimestamp, previousMsgHash} ->               case agentMessage of-                HELLO verifyKey _ -> helloMsg verifyKey msgBody-                REPLY qInfo -> replyMsg qInfo+                HELLO verifyKey _ -> helloMsg verifyKey msgBody >> sendAck c rq+                REPLY qInfo -> replyMsg qInfo >> sendAck c rq                 A_MSG body -> agentClientMsg previousMsgHash (senderMsgId, senderTimestamp) (srvMsgId, srvTs) body msgHash-          sendAck c rq-          return ()         SMP.END -> do-          removeSubscription c connAlias+          removeSubscription c connId           logServer "<--" c srv rId "END"           notify END         _ -> do@@ -311,22 +524,25 @@           notify . ERR $ BROKER UNEXPECTED       where         notify :: ACommand 'Agent -> m ()-        notify msg = atomically $ writeTBQueue sndQ ("", connAlias, msg)+        notify msg = atomically $ writeTBQueue subQ ("", connId, msg)          prohibited :: m ()         prohibited = notify . ERR $ AGENT A_PROHIBITED -        smpConfirmation :: SenderPublicKey -> m ()-        smpConfirmation senderKey = do+        smpConfirmation :: SenderPublicKey -> ConnInfo -> m ()+        smpConfirmation senderKey cInfo = do           logServer "<--" c srv rId "MSG <KEY>"           case status of-            New -> do-              -- TODO currently it automatically allows whoever sends the confirmation-              -- Commands CONF and LET are not supported in v0.2-              withStore $ setRcvQueueStatus st rq Confirmed-              -- TODO update sender key in the store?-              secureQueue c rq senderKey-              withStore $ setRcvQueueStatus st rq Secured+            New -> case cType of+              SCRcv -> do+                g <- asks idsDrg+                let newConfirmation = NewConfirmation {connId, senderKey, senderConnInfo = cInfo}+                confId <- withStore $ \st -> createConfirmation st g newConfirmation+                notify $ REQ confId cInfo+              SCDuplex -> do+                notify $ INFO cInfo+                processConfirmation c rq senderKey+              _ -> prohibited             _ -> prohibited          helloMsg :: SenderPublicKey -> ByteString -> m ()@@ -336,9 +552,9 @@             Active -> prohibited             _ -> do               void $ verifyMessage (Just verifyKey) msgBody-              withStore $ setRcvQueueActive st rq verifyKey+              withStore $ \st -> setRcvQueueActive st rq verifyKey               case cType of-                SCDuplex -> notify CON+                SCDuplex -> notifyConnected c connId                 _ -> pure ()          replyMsg :: SMPQueueInfo -> m ()@@ -346,42 +562,26 @@           logServer "<--" c srv rId "MSG <REPLY>"           case cType of             SCRcv -> do-              (sq, senderKey, verifyKey) <- newSendQueue qInfo connAlias-              withStore $ upgradeRcvConnToDuplex st connAlias sq-              connectToSendQueue c st sq senderKey verifyKey-              notify CON+              AcceptedConfirmation {ownConnInfo} <- withStore (`getAcceptedConfirmation` connId)+              (sq, senderKey, verifyKey) <- newSndQueue qInfo+              withStore $ \st -> upgradeRcvConnToDuplex st connId sq+              confirmQueue c sq senderKey ownConnInfo+              withStore (`removeConfirmations` connId)+              cfg <- asks config+              activateQueueInitiating c connId sq verifyKey $ retryInterval cfg             _ -> prohibited          agentClientMsg :: PrevRcvMsgHash -> (ExternalSndId, ExternalSndTs) -> (BrokerId, BrokerTs) -> MsgBody -> MsgHash -> m ()-        agentClientMsg receivedPrevMsgHash senderMeta brokerMeta msgBody msgHash = do+        agentClientMsg externalPrevSndHash sender broker msgBody internalHash = do           logServer "<--" c srv rId "MSG <MSG>"-          case status of-            Active -> do-              internalTs <- liftIO getCurrentTime-              (internalId, internalRcvId, prevExtSndId, prevRcvMsgHash) <- withStore $ updateRcvIds st rq-              let msgIntegrity = checkMsgIntegrity prevExtSndId (fst senderMeta) prevRcvMsgHash receivedPrevMsgHash-              withStore $-                createRcvMsg st rq $-                  RcvMsgData-                    { internalId,-                      internalRcvId,-                      internalTs,-                      senderMeta,-                      brokerMeta,-                      msgBody,-                      internalHash = msgHash,-                      externalPrevSndHash = receivedPrevMsgHash,-                      msgIntegrity-                    }-              notify-                MSG-                  { recipientMeta = (unId internalId, internalTs),-                    senderMeta,-                    brokerMeta,-                    msgBody,-                    msgIntegrity-                  }-            _ -> prohibited+          internalTs <- liftIO getCurrentTime+          (internalId, internalRcvId, prevExtSndId, prevRcvMsgHash) <- withStore (`updateRcvIds` connId)+          let integrity = checkMsgIntegrity prevExtSndId (fst sender) prevRcvMsgHash externalPrevSndHash+              recipient = (unId internalId, internalTs)+              msgMeta = MsgMeta {integrity, recipient, sender, broker}+              rcvMsg = RcvMsgData {..}+          withStore $ \st -> createRcvMsg st connId rcvMsg+          notify $ MSG msgMeta msgBody          checkMsgIntegrity :: PrevExternalSndId -> ExternalSndId -> PrevRcvMsgHash -> ByteString -> MsgIntegrity         checkMsgIntegrity prevExtSndId extSndId internalPrevMsgHash receivedPrevMsgHash@@ -392,16 +592,39 @@           | internalPrevMsgHash /= receivedPrevMsgHash = MsgError MsgBadHash           | otherwise = MsgError MsgDuplicate -- this case is not possible -connectToSendQueue :: AgentMonad m => AgentClient -> SQLiteStore -> SndQueue -> SenderPublicKey -> VerificationKey -> m ()-connectToSendQueue c st sq senderKey verifyKey = do-  sendConfirmation c sq senderKey-  withStore $ setSndQueueStatus st sq Confirmed-  sendHello c sq verifyKey-  withStore $ setSndQueueStatus st sq Active+confirmQueue :: AgentMonad m => AgentClient -> SndQueue -> SenderPublicKey -> ConnInfo -> m ()+confirmQueue c sq senderKey cInfo = do+  sendConfirmation c sq senderKey cInfo+  withStore $ \st -> setSndQueueStatus st sq Confirmed -newSendQueue ::-  (MonadUnliftIO m, MonadReader Env m) => SMPQueueInfo -> ConnAlias -> m (SndQueue, SenderPublicKey, VerificationKey)-newSendQueue (SMPQueueInfo smpServer senderId encryptKey) connAlias = do+activateQueueInitiating :: AgentMonad m => AgentClient -> ConnId -> SndQueue -> VerificationKey -> RetryInterval -> m ()+activateQueueInitiating c connId sq verifyKey retryInterval =+  activateQueue c connId sq verifyKey retryInterval $ notifyConnected c connId++activateQueue :: forall m. AgentMonad m => AgentClient -> ConnId -> SndQueue -> VerificationKey -> RetryInterval -> m () -> m ()+activateQueue c connId sq verifyKey retryInterval afterActivation =+  getActivation c connId >>= \case+    Nothing -> async runActivation >>= addActivation c connId+    Just _ -> pure ()+  where+    runActivation :: m ()+    runActivation = do+      sendHello c sq verifyKey retryInterval+      withStore $ \st -> setSndQueueStatus st sq Active+      removeActivation c connId+      removeVerificationKey+      afterActivation+    removeVerificationKey :: m ()+    removeVerificationKey =+      let safeSignKey = C.removePublicKey $ signKey sq+       in withStore $ \st -> updateSignKey st sq safeSignKey++notifyConnected :: AgentMonad m => AgentClient -> ConnId -> m ()+notifyConnected c connId = atomically $ writeTBQueue (subQ c) ("", connId, CON)++newSndQueue ::+  (MonadUnliftIO m, MonadReader Env m) => SMPQueueInfo -> m (SndQueue, SenderPublicKey, VerificationKey)+newSndQueue (SMPQueueInfo smpServer senderId encryptKey) = do   size <- asks $ rsaKeySize . config   (senderKey, sndPrivateKey) <- liftIO $ C.generateKeyPair size   (verifyKey, signKey) <- liftIO $ C.generateKeyPair size@@ -409,7 +632,6 @@         SndQueue           { server = smpServer,             sndId = senderId,-            connAlias,             sndPrivateKey,             encryptKey,             signKey,
src/Simplex/Messaging/Agent/Client.hs view
@@ -11,11 +11,13 @@   ( AgentClient (..),     newAgentClient,     AgentMonad,-    getSMPServerClient,-    closeSMPServerClients,-    newReceiveQueue,+    withAgentLock,+    closeAgentClient,+    newRcvQueue,     subscribeQueue,+    addSubscription,     sendConfirmation,+    RetryInterval (..),     sendHello,     secureQueue,     sendAgentMessage,@@ -27,9 +29,14 @@     logServer,     removeSubscription,     cryptoError,+    addActivation,+    getActivation,+    removeActivation,   ) where +import Control.Concurrent.Async (Async, async, uninterruptibleCancel)+import Control.Concurrent.STM (stateTVar) import Control.Logger.Simple import Control.Monad.Except import Control.Monad.IO.Unlift@@ -40,44 +47,62 @@ import qualified Data.ByteString.Char8 as B import Data.Map.Strict (Map) import qualified Data.Map.Strict as M+import Data.Maybe (isNothing) import Data.Set (Set) import qualified Data.Set as S import Data.Text.Encoding import Data.Time.Clock import Simplex.Messaging.Agent.Env.SQLite import Simplex.Messaging.Agent.Protocol+import Simplex.Messaging.Agent.RetryInterval import Simplex.Messaging.Agent.Store import Simplex.Messaging.Client import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgBody, QueueId, SenderPublicKey) import Simplex.Messaging.Util (bshow, liftEitherError, liftError)-import UnliftIO.Concurrent import UnliftIO.Exception (IOException) import qualified UnliftIO.Exception as E import UnliftIO.STM  data AgentClient = AgentClient   { rcvQ :: TBQueue (ATransmission 'Client),-    sndQ :: TBQueue (ATransmission 'Agent),+    subQ :: TBQueue (ATransmission 'Agent),     msgQ :: TBQueue SMPServerTransmission,     smpClients :: TVar (Map SMPServer SMPClient),-    subscrSrvrs :: TVar (Map SMPServer (Set ConnAlias)),-    subscrConns :: TVar (Map ConnAlias SMPServer),-    clientId :: Int+    subscrSrvrs :: TVar (Map SMPServer (Map ConnId RcvQueue)),+    subscrConns :: TVar (Map ConnId SMPServer),+    activations :: TVar (Map ConnId (Async ())), -- activations of send queues in progress+    connMsgQueues :: TVar (Map ConnId (TQueue PendingMsg)),+    connMsgDeliveries :: TVar (Map ConnId (Async ())),+    srvMsgQueues :: TVar (Map SMPServer (TQueue PendingMsg)),+    srvMsgDeliveries :: TVar (Map SMPServer (Async ())),+    reconnections :: TVar [Async ()],+    clientId :: Int,+    agentEnv :: Env,+    smpSubscriber :: Async (),+    lock :: TMVar ()   } -newAgentClient :: TVar Int -> AgentConfig -> STM AgentClient-newAgentClient cc AgentConfig {tbqSize} = do-  rcvQ <- newTBQueue tbqSize-  sndQ <- newTBQueue tbqSize-  msgQ <- newTBQueue tbqSize+newAgentClient :: Env -> STM AgentClient+newAgentClient agentEnv = do+  let qSize = tbqSize $ config agentEnv+  rcvQ <- newTBQueue qSize+  subQ <- newTBQueue qSize+  msgQ <- newTBQueue qSize   smpClients <- newTVar M.empty   subscrSrvrs <- newTVar M.empty   subscrConns <- newTVar M.empty-  clientId <- (+ 1) <$> readTVar cc-  writeTVar cc clientId-  return AgentClient {rcvQ, sndQ, msgQ, smpClients, subscrSrvrs, subscrConns, clientId}+  activations <- newTVar M.empty+  connMsgQueues <- newTVar M.empty+  connMsgDeliveries <- newTVar M.empty+  srvMsgQueues <- newTVar M.empty+  srvMsgDeliveries <- newTVar M.empty+  reconnections <- newTVar []+  clientId <- stateTVar (clientCounter agentEnv) $ \i -> (i + 1, i + 1)+  lock <- newTMVar ()+  return AgentClient {rcvQ, subQ, msgQ, smpClients, subscrSrvrs, subscrConns, activations, connMsgQueues, connMsgDeliveries, srvMsgQueues, srvMsgDeliveries, reconnections, clientId, agentEnv, smpSubscriber = undefined, lock} +-- | Agent monad with MonadReader Env and MonadError AgentErrorType type AgentMonad m = (MonadUnliftIO m, MonadReader Env m, MonadError AgentErrorType m)  getSMPServerClient :: forall m. AgentMonad m => AgentClient -> SMPServer -> m SMPClient@@ -95,34 +120,77 @@     connectClient :: m SMPClient     connectClient = do       cfg <- asks $ smpCfg . config-      liftEitherError smpClientError (getSMPClient srv cfg msgQ clientDisconnected)+      u <- askUnliftIO+      liftEitherError smpClientError (getSMPClient srv cfg msgQ $ clientDisconnected u)         `E.catch` internalError       where         internalError :: IOException -> m SMPClient         internalError = throwError . INTERNAL . show -    clientDisconnected :: IO ()-    clientDisconnected = do-      removeSubs >>= mapM_ (mapM_ notifySub)+    clientDisconnected :: UnliftIO m -> IO ()+    clientDisconnected u = do+      removeClientSubs >>= (`forM_` serverDown u)       logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv -    removeSubs :: IO (Maybe (Set ConnAlias))-    removeSubs = atomically $ do+    removeClientSubs :: IO (Maybe (Map ConnId RcvQueue))+    removeClientSubs = atomically $ do       modifyTVar smpClients $ M.delete srv       cs <- M.lookup srv <$> readTVar (subscrSrvrs c)       modifyTVar (subscrSrvrs c) $ M.delete srv-      modifyTVar (subscrConns c) $ maybe id deleteKeys cs+      modifyTVar (subscrConns c) $ maybe id (deleteKeys . M.keysSet) cs       return cs       where         deleteKeys :: Ord k => Set k -> Map k a -> Map k a         deleteKeys ks m = S.foldr' M.delete m ks -    notifySub :: ConnAlias -> IO ()-    notifySub connAlias = atomically $ writeTBQueue (sndQ c) ("", connAlias, END)+    serverDown :: UnliftIO m -> Map ConnId RcvQueue -> IO ()+    serverDown u cs = unless (M.null cs) $ do+      mapM_ (notifySub DOWN) $ M.keysSet cs+      a <- async . unliftIO u $ tryReconnectClient cs+      atomically $ modifyTVar (reconnections c) (a :) -closeSMPServerClients :: MonadUnliftIO m => AgentClient -> m ()-closeSMPServerClients c = liftIO $ readTVarIO (smpClients c) >>= mapM_ closeSMPClient+    tryReconnectClient :: Map ConnId RcvQueue -> m ()+    tryReconnectClient cs = do+      ri <- asks $ reconnectInterval . config+      withRetryInterval ri $ \loop ->+        reconnectClient cs `catchError` const loop +    reconnectClient :: Map ConnId RcvQueue -> m ()+    reconnectClient cs = do+      withAgentLock c . withSMP c srv $ \smp -> do+        subs <- readTVarIO $ subscrConns c+        forM_ (M.toList cs) $ \(connId, rq@RcvQueue {rcvPrivateKey, rcvId}) ->+          when (isNothing $ M.lookup connId subs) $ do+            subscribeSMPQueue smp rcvPrivateKey rcvId+              `catchError` \case+                SMPServerError e -> liftIO $ notifySub (ERR $ SMP e) connId+                e -> throwError e+            addSubscription c rq connId+            liftIO $ notifySub UP connId++    notifySub :: ACommand 'Agent -> ConnId -> IO ()+    notifySub cmd connId = atomically $ writeTBQueue (subQ c) ("", connId, cmd)++closeAgentClient :: MonadUnliftIO m => AgentClient -> m ()+closeAgentClient c = liftIO $ do+  closeSMPServerClients c+  cancelActions $ activations c+  cancelActions $ reconnections c+  cancelActions $ connMsgDeliveries c+  cancelActions $ srvMsgDeliveries c++closeSMPServerClients :: AgentClient -> IO ()+closeSMPServerClients c = readTVarIO (smpClients c) >>= mapM_ closeSMPClient++cancelActions :: Foldable f => TVar (f (Async ())) -> IO ()+cancelActions as = readTVarIO as >>= mapM_ uninterruptibleCancel++withAgentLock :: MonadUnliftIO m => AgentClient -> m a -> m a+withAgentLock AgentClient {lock} =+  E.bracket_+    (void . atomically $ takeTMVar lock)+    (atomically $ putTMVar lock ())+ withSMP_ :: forall a m. AgentMonad m => AgentClient -> SMPServer -> (SMPClient -> m a) -> m a withSMP_ c srv action =   (getSMPServerClient c srv >>= action) `catchError` logServerError@@ -158,8 +226,8 @@   SMPTransportError e -> BROKER $ TRANSPORT e   e -> INTERNAL $ show e -newReceiveQueue :: AgentMonad m => AgentClient -> SMPServer -> ConnAlias -> m (RcvQueue, SMPQueueInfo)-newReceiveQueue c srv connAlias = do+newRcvQueue :: AgentMonad m => AgentClient -> SMPServer -> m (RcvQueue, SMPQueueInfo)+newRcvQueue c srv = do   size <- asks $ rsaKeySize . config   (recipientKey, rcvPrivateKey) <- liftIO $ C.generateKeyPair size   logServer "-->" c srv "" "NEW"@@ -170,45 +238,51 @@         RcvQueue           { server = srv,             rcvId,-            connAlias,             rcvPrivateKey,             sndId = Just sId,-            sndKey = Nothing,             decryptKey,             verifyKey = Nothing,             status = New           }-  addSubscription c rq connAlias   return (rq, SMPQueueInfo srv sId encryptKey) -subscribeQueue :: AgentMonad m => AgentClient -> RcvQueue -> ConnAlias -> m ()-subscribeQueue c rq@RcvQueue {server, rcvPrivateKey, rcvId} connAlias = do+subscribeQueue :: AgentMonad m => AgentClient -> RcvQueue -> ConnId -> m ()+subscribeQueue c rq@RcvQueue {server, rcvPrivateKey, rcvId} connId = do   withLogSMP c server rcvId "SUB" $ \smp ->     subscribeSMPQueue smp rcvPrivateKey rcvId-  addSubscription c rq connAlias+  addSubscription c rq connId -addSubscription :: MonadUnliftIO m => AgentClient -> RcvQueue -> ConnAlias -> m ()-addSubscription c RcvQueue {server} connAlias = atomically $ do-  modifyTVar (subscrConns c) $ M.insert connAlias server+addSubscription :: MonadUnliftIO m => AgentClient -> RcvQueue -> ConnId -> m ()+addSubscription c rq@RcvQueue {server} connId = atomically $ do+  modifyTVar (subscrConns c) $ M.insert connId server   modifyTVar (subscrSrvrs c) $ M.alter (Just . addSub) server   where-    addSub :: Maybe (Set ConnAlias) -> Set ConnAlias-    addSub (Just cs) = S.insert connAlias cs-    addSub _ = S.singleton connAlias+    addSub :: Maybe (Map ConnId RcvQueue) -> Map ConnId RcvQueue+    addSub (Just cs) = M.insert connId rq cs+    addSub _ = M.singleton connId rq -removeSubscription :: AgentMonad m => AgentClient -> ConnAlias -> m ()-removeSubscription AgentClient {subscrConns, subscrSrvrs} connAlias = atomically $ do+removeSubscription :: AgentMonad m => AgentClient -> ConnId -> m ()+removeSubscription AgentClient {subscrConns, subscrSrvrs} connId = atomically $ do   cs <- readTVar subscrConns-  writeTVar subscrConns $ M.delete connAlias cs+  writeTVar subscrConns $ M.delete connId cs   mapM_     (modifyTVar subscrSrvrs . M.alter (>>= delSub))-    (M.lookup connAlias cs)+    (M.lookup connId cs)   where-    delSub :: Set ConnAlias -> Maybe (Set ConnAlias)+    delSub :: Map ConnId RcvQueue -> Maybe (Map ConnId RcvQueue)     delSub cs =-      let cs' = S.delete connAlias cs-       in if S.null cs' then Nothing else Just cs'+      let cs' = M.delete connId cs+       in if M.null cs' then Nothing else Just cs' +addActivation :: MonadUnliftIO m => AgentClient -> ConnId -> Async () -> m ()+addActivation c connId a = atomically . modifyTVar (activations c) $ M.insert connId a++getActivation :: MonadUnliftIO m => AgentClient -> ConnId -> m (Maybe (Async ()))+getActivation c connId = M.lookup connId <$> readTVarIO (activations c)++removeActivation :: MonadUnliftIO m => AgentClient -> ConnId -> m ()+removeActivation c connId = atomically . modifyTVar (activations c) $ M.delete connId+ logServer :: AgentMonad m => ByteString -> AgentClient -> SMPServer -> QueueId -> ByteString -> m () logServer dir AgentClient {clientId} srv qId cmdStr =   logInfo . decodeUtf8 $ B.unwords ["A", "(" <> bshow clientId <> ")", dir, showServer srv, ":", logSecret qId, cmdStr]@@ -219,20 +293,23 @@ logSecret :: ByteString -> ByteString logSecret bs = encode $ B.take 3 bs -sendConfirmation :: forall m. AgentMonad m => AgentClient -> SndQueue -> SenderPublicKey -> m ()-sendConfirmation c sq@SndQueue {server, sndId} senderKey =+sendConfirmation :: forall m. AgentMonad m => AgentClient -> SndQueue -> SenderPublicKey -> ConnInfo -> m ()+sendConfirmation c sq@SndQueue {server, sndId} senderKey cInfo =   withLogSMP_ c server sndId "SEND <KEY>" $ \smp -> do     msg <- mkConfirmation smp     liftSMP $ sendSMPMessage smp Nothing sndId msg   where     mkConfirmation :: SMPClient -> m MsgBody-    mkConfirmation smp = encryptAndSign smp sq . serializeSMPMessage $ SMPConfirmation senderKey+    mkConfirmation smp = encryptAndSign smp sq . serializeSMPMessage $ SMPConfirmation senderKey cInfo -sendHello :: forall m. AgentMonad m => AgentClient -> SndQueue -> VerificationKey -> m ()-sendHello c sq@SndQueue {server, sndId, sndPrivateKey} verifyKey =+sendHello :: forall m. AgentMonad m => AgentClient -> SndQueue -> VerificationKey -> RetryInterval -> m ()+sendHello c sq@SndQueue {server, sndId, sndPrivateKey} verifyKey ri =   withLogSMP_ c server sndId "SEND <HELLO> (retrying)" $ \smp -> do     msg <- mkHello smp $ AckMode On-    liftSMP $ send 8 100000 msg smp+    liftSMP . withRetryInterval ri $ \loop ->+      sendSMPMessage smp (Just sndPrivateKey) sndId msg `catchE` \case+        SMPServerError AUTH -> loop+        e -> throwE e   where     mkHello :: SMPClient -> AckMode -> m ByteString     mkHello smp ackMode = do@@ -244,15 +321,6 @@             previousMsgHash = "",             agentMessage = HELLO verifyKey ackMode           }--    send :: Int -> Int -> ByteString -> SMPClient -> ExceptT SMPClientError IO ()-    send 0 _ _ _ = throwE $ SMPServerError AUTH-    send retry delay msg smp =-      sendSMPMessage smp (Just sndPrivateKey) sndId msg `catchE` \case-        SMPServerError AUTH -> do-          threadDelay delay-          send (retry - 1) (delay * 3 `div` 2) msg smp-        e -> throwE e  secureQueue :: AgentMonad m => AgentClient -> RcvQueue -> SenderPublicKey -> m () secureQueue c RcvQueue {server, rcvId, rcvPrivateKey} senderKey =
src/Simplex/Messaging/Agent/Env/SQLite.hs view
@@ -1,6 +1,7 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE NumericUnderscores #-} {-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-}  module Simplex.Messaging.Agent.Env.SQLite where@@ -11,7 +12,9 @@ import Network.Socket import Numeric.Natural import Simplex.Messaging.Agent.Protocol (SMPServer)+import Simplex.Messaging.Agent.RetryInterval import Simplex.Messaging.Agent.Store.SQLite+import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations import Simplex.Messaging.Client import System.Random (StdGen, newStdGen) import UnliftIO.STM@@ -23,11 +26,43 @@     connIdBytes :: Int,     tbqSize :: Natural,     dbFile :: FilePath,-    smpCfg :: SMPClientConfig+    dbPoolSize :: Int,+    smpCfg :: SMPClientConfig,+    retryInterval :: RetryInterval,+    reconnectInterval :: RetryInterval   } +minute :: Int+minute = 60_000_000++defaultAgentConfig :: AgentConfig+defaultAgentConfig =+  AgentConfig+    { tcpPort = "5224",+      smpServers = undefined,+      rsaKeySize = 2048 `div` 8,+      connIdBytes = 12,+      tbqSize = 16,+      dbFile = "smp-agent.db",+      dbPoolSize = 4,+      smpCfg = smpDefaultConfig,+      retryInterval =+        RetryInterval+          { initialInterval = 1_000_000,+            increaseAfter = minute,+            maxInterval = 10 * minute+          },+      reconnectInterval =+        RetryInterval+          { initialInterval = 1_000_000,+            increaseAfter = 10_000_000,+            maxInterval = 10_000_000+          }+    }+ data Env = Env   { config :: AgentConfig,+    store :: SQLiteStore,     idsDrg :: TVar ChaChaDRG,     clientCounter :: TVar Int,     reservedMsgSize :: Int,@@ -35,15 +70,15 @@   }  newSMPAgentEnv :: (MonadUnliftIO m, MonadRandom m) => AgentConfig -> m Env-newSMPAgentEnv config = do+newSMPAgentEnv cfg = do   idsDrg <- newTVarIO =<< drgNew-  _ <- createSQLiteStore $ dbFile config+  store <- liftIO $ createSQLiteStore (dbFile cfg) (dbPoolSize cfg) Migrations.app   clientCounter <- newTVarIO 0   randomServer <- newTVarIO =<< liftIO newStdGen-  return Env {config, idsDrg, clientCounter, reservedMsgSize, randomServer}+  return Env {config = cfg, store, idsDrg, clientCounter, reservedMsgSize, randomServer}   where     -- 1st rsaKeySize is used by the RSA signature in each command,     -- 2nd - by encrypted message body header     -- 3rd - by message signature     -- smpCommandSize - is the estimated max size for SMP command, queueId, corrId-    reservedMsgSize = 3 * rsaKeySize config + smpCommandSize (smpCfg config)+    reservedMsgSize = 3 * rsaKeySize cfg + smpCommandSize (smpCfg cfg)
src/Simplex/Messaging/Agent/Protocol.hs view
@@ -1,14 +1,17 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE DeriveAnyClass #-} {-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE GADTs #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE PolyKinds #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-} {-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-}  -- |@@ -25,9 +28,12 @@ -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/agent-protocol.md module Simplex.Messaging.Agent.Protocol   ( -- * SMP agent protocol types+    ConnInfo,     ACommand (..),     AParty (..),     SAParty (..),+    MsgHash,+    MsgMeta (..),     SMPMessage (..),     AMessage (..),     SMPServer (..),@@ -40,8 +46,10 @@     ATransmission,     ATransmissionOrError,     ARawTransmission,-    ConnAlias,-    ReplyMode (..),+    ConnId,+    ConfirmationId,+    IntroId,+    InvitationId,     AckMode (..),     OnOff (..),     MsgIntegrity (..),@@ -51,6 +59,8 @@     VerificationKey,     EncryptionKey,     DecryptionKey,+    ACorrId,+    AgentMsgId,      -- * Parse and serialize     serializeCommand,@@ -91,12 +101,11 @@ import Data.Typeable () import GHC.Generics (Generic) import Generic.Random (genericArbitraryU)-import Network.Socket+import Network.Socket (HostName, ServiceName) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Parsers import Simplex.Messaging.Protocol-  ( CorrId (..),-    ErrorType,+  ( ErrorType,     MsgBody,     MsgId,     SenderPublicKey,@@ -112,11 +121,13 @@ type ARawTransmission = (ByteString, ByteString, ByteString)  -- | Parsed SMP agent protocol transmission.-type ATransmission p = (CorrId, ConnAlias, ACommand p)+type ATransmission p = (ACorrId, ConnId, ACommand p)  -- | SMP agent protocol transmission or transmission error.-type ATransmissionOrError p = (CorrId, ConnAlias, Either AgentErrorType (ACommand p))+type ATransmissionOrError p = (ACorrId, ConnId, Either AgentErrorType (ACommand p)) +type ACorrId = ByteString+ -- | SMP agent protocol participants. data AParty = Agent | Client   deriving (Eq, Show)@@ -139,31 +150,29 @@  deriving instance Show ACmd +type ConnInfo = ByteString+ -- | Parameterized type for SMP agent protocol commands and responses from all participants. data ACommand (p :: AParty) where   NEW :: ACommand Client -- response INV   INV :: SMPQueueInfo -> ACommand Agent-  JOIN :: SMPQueueInfo -> ReplyMode -> ACommand Client -- response OK+  JOIN :: SMPQueueInfo -> ConnInfo -> ACommand Client -- response OK+  REQ :: ConfirmationId -> ConnInfo -> ACommand Agent -- ConnInfo is from sender+  ACPT :: ConfirmationId -> ConnInfo -> ACommand Client -- ConnInfo is from client+  INFO :: ConnInfo -> ACommand Agent   CON :: ACommand Agent -- notification that connection is established-  -- TODO currently it automatically allows whoever sends the confirmation-  -- CONF :: OtherPartyId -> ACommand Agent-  -- LET :: OtherPartyId -> ACommand Client   SUB :: ACommand Client-  SUBALL :: ACommand Client -- TODO should be moved to chat protocol - hack for subscribing to all   END :: ACommand Agent+  DOWN :: ACommand Agent+  UP :: ACommand Agent   -- QST :: QueueDirection -> ACommand Client   -- STAT :: QueueDirection -> Maybe QueueStatus -> Maybe SubMode -> ACommand Agent   SEND :: MsgBody -> ACommand Client+  MID :: AgentMsgId -> ACommand Agent   SENT :: AgentMsgId -> ACommand Agent-  MSG ::-    { recipientMeta :: (AgentMsgId, UTCTime),-      brokerMeta :: (MsgId, UTCTime),-      senderMeta :: (AgentMsgId, UTCTime),-      msgIntegrity :: MsgIntegrity,-      msgBody :: MsgBody-    } ->-    ACommand Agent-  -- ACK :: AgentMsgId -> ACommand Client+  MERR :: AgentMsgId -> AgentErrorType -> ACommand Agent+  MSG :: MsgMeta -> MsgBody -> ACommand Agent+  ACK :: AgentMsgId -> ACommand Client   -- RCVD :: AgentMsgId -> ACommand Agent   OFF :: ACommand Client   DEL :: ACommand Client@@ -174,11 +183,27 @@  deriving instance Show (ACommand p) +type MsgHash = ByteString++-- | Agent message metadata sent to the client+data MsgMeta = MsgMeta+  { integrity :: MsgIntegrity,+    recipient :: (AgentMsgId, UTCTime),+    broker :: (MsgId, UTCTime),+    sender :: (AgentMsgId, UTCTime)+  }+  deriving (Eq, Show)+ -- | SMP message formats. data SMPMessage   = -- | SMP confirmation     -- (see <https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#send-message SMP protocol>)-    SMPConfirmation SenderPublicKey+    SMPConfirmation+      { -- | sender's public key to use for authentication of sender's commands at the recepient's server+        senderKey :: SenderPublicKey,+        -- | sender's information to be associated with the connection, e.g. sender's profile information+        connInfo :: ConnInfo+      }   | -- | Agent message header and envelope for client messages     -- (see <https://github.com/simplex-chat/simplexmq/blob/master/protocol/agent-protocol.md#messages-between-smp-agents SMP agent protocol>)     SMPMessage@@ -187,7 +212,7 @@         -- | timestamp from the sending agent         senderTimestamp :: SenderTimestamp,         -- | digest of the previous message-        previousMsgHash :: ByteString,+        previousMsgHash :: MsgHash,         -- | messages sent between agents once queue is secured         agentMessage :: AMessage       }@@ -210,12 +235,10 @@ parseSMPMessage = parse (smpMessageP <* A.endOfLine) $ AGENT A_MESSAGE   where     smpMessageP :: Parser SMPMessage-    smpMessageP =-      smpConfirmationP <* A.endOfLine-        <|> A.endOfLine *> smpClientMessageP+    smpMessageP = A.endOfLine *> smpClientMessageP <|> smpConfirmationP      smpConfirmationP :: Parser SMPMessage-    smpConfirmationP = SMPConfirmation <$> ("KEY " *> C.pubKeyP <* A.endOfLine)+    smpConfirmationP = "KEY " *> (SMPConfirmation <$> C.pubKeyP <* A.endOfLine <* A.endOfLine <*> binaryBodyP <* A.endOfLine)      smpClientMessageP :: Parser SMPMessage     smpClientMessageP =@@ -230,7 +253,7 @@ -- | Serialize SMP message. serializeSMPMessage :: SMPMessage -> ByteString serializeSMPMessage = \case-  SMPConfirmation sKey -> smpMessage ("KEY " <> C.serializePubKey sKey) "" ""+  SMPConfirmation sKey cInfo -> smpMessage ("KEY " <> C.serializePubKey sKey) "" (serializeBinary cInfo) <> "\n"   SMPMessage {senderMsgId, senderTimestamp, previousMsgHash, agentMessage} ->     let header = messageHeader senderMsgId senderTimestamp previousMsgHash         body = serializeAgentMessage agentMessage@@ -248,9 +271,7 @@   where     hello = HELLO <$> C.pubKeyP <*> ackMode     reply = REPLY <$> smpQueueInfoP-    a_msg = do-      size :: Int <- A.decimal <* A.endOfLine-      A_MSG <$> A.take size <* A.endOfLine+    a_msg = A_MSG <$> binaryBodyP <* A.endOfLine     ackMode = AckMode <$> (" NO_ACK" $> Off <|> pure On)  -- | SMP queue information parser.@@ -262,7 +283,7 @@ smpServerP :: Parser SMPServer smpServerP = SMPServer <$> server <*> optional port <*> optional kHash   where-    server = B.unpack <$> A.takeWhile1 (A.notInClass ":# ")+    server = B.unpack <$> A.takeWhile1 (A.notInClass ":#,; ")     port = A.char ':' *> (B.unpack <$> A.takeWhile1 A.isDigit)     kHash = C.KeyHash <$> (A.char '#' *> base64P) @@ -270,7 +291,7 @@ serializeAgentMessage = \case   HELLO verifyKey ackMode -> "HELLO " <> C.serializePubKey verifyKey <> if ackMode == AckMode Off then " NO_ACK" else ""   REPLY qInfo -> "REPLY " <> serializeSmpQueueInfo qInfo-  A_MSG body -> "MSG " <> serializeMsg body <> "\n"+  A_MSG body -> "MSG " <> serializeBinary body <> "\n"  -- | Serialize SMP queue information that is sent out-of-band. serializeSmpQueueInfo :: SMPQueueInfo -> ByteString@@ -291,11 +312,17 @@   deriving (Eq, Ord, Show)  instance IsString SMPServer where-  fromString = parseString . parseAll $ smpServerP+  fromString = parseString $ parseAll smpServerP  -- | SMP agent connection alias.-type ConnAlias = ByteString+type ConnId = ByteString +type ConfirmationId = ByteString++type IntroId = ByteString++type InvitationId = ByteString+ -- | Connection modes. data OnOff = On | Off deriving (Eq, Show, Read) @@ -308,9 +335,6 @@ data SMPQueueInfo = SMPQueueInfo SMPServer SMP.SenderId EncryptionKey   deriving (Eq, Show) --- | Connection reply mode (used in JOIN command).-newtype ReplyMode = ReplyMode OnOff deriving (Eq, Show)- -- | Public key used to E2E encrypt SMP messages. type EncryptionKey = C.PublicKey @@ -318,7 +342,7 @@ type DecryptionKey = C.SafePrivateKey  -- | Private key used to sign SMP commands-type SignatureKey = C.SafePrivateKey+type SignatureKey = C.APrivateKey  -- | Public key used by SMP server to authorize (verify) SMP commands. type VerificationKey = C.PublicKey@@ -369,11 +393,11 @@  -- | SMP agent protocol command or response error. data CommandErrorType-  = -- | command is prohibited+  = -- | command is prohibited in this context     PROHIBITED   | -- | command syntax is invalid     SYNTAX-  | -- | connection alias is required with this command+  | -- | entity ID is required with this command     NO_CONN   | -- | message size is not correct (no terminating space)     SIZE@@ -384,7 +408,7 @@ -- | Connection error. data ConnectionErrorType   = -- | connection alias is not in the database-    UNKNOWN+    NOT_FOUND   | -- | connection alias already exists     DUPLICATE   | -- | connection is simplex, but operation requires another queue@@ -427,18 +451,25 @@  instance Arbitrary SMPAgentError where arbitrary = genericArbitraryU --- | AMP agent command and response parser+-- | SMP agent command and response parser commandP :: Parser ACmd commandP =   "NEW" $> ACmd SClient NEW     <|> "INV " *> invResp     <|> "JOIN " *> joinCmd+    <|> "REQ " *> reqCmd+    <|> "ACPT " *> acptCmd+    <|> "INFO " *> infoCmd     <|> "SUB" $> ACmd SClient SUB-    <|> "SUBALL" $> ACmd SClient SUBALL -- TODO remove - hack for subscribing to all     <|> "END" $> ACmd SAgent END+    <|> "DOWN" $> ACmd SAgent DOWN+    <|> "UP" $> ACmd SAgent UP     <|> "SEND " *> sendCmd+    <|> "MID " *> msgIdResp     <|> "SENT " *> sentResp+    <|> "MERR " *> msgErrResp     <|> "MSG " *> message+    <|> "ACK " *> ackCmd     <|> "OFF" $> ACmd SClient OFF     <|> "DEL" $> ACmd SClient DEL     <|> "ERR " *> agentError@@ -446,18 +477,23 @@     <|> "OK" $> ACmd SAgent OK   where     invResp = ACmd SAgent . INV <$> smpQueueInfoP-    joinCmd = ACmd SClient <$> (JOIN <$> smpQueueInfoP <*> replyMode)+    joinCmd = ACmd SClient <$> (JOIN <$> smpQueueInfoP <* A.space <*> A.takeByteString)+    reqCmd = ACmd SAgent <$> (REQ <$> A.takeTill (== ' ') <* A.space <*> A.takeByteString)+    acptCmd = ACmd SClient <$> (ACPT <$> A.takeTill (== ' ') <* A.space <*> A.takeByteString)+    infoCmd = ACmd SAgent . INFO <$> A.takeByteString     sendCmd = ACmd SClient . SEND <$> A.takeByteString+    msgIdResp = ACmd SAgent . MID <$> A.decimal     sentResp = ACmd SAgent . SENT <$> A.decimal-    message = do-      msgIntegrity <- msgIntegrityP <* A.space-      recipientMeta <- "R=" *> partyMeta A.decimal-      brokerMeta <- "B=" *> partyMeta base64P-      senderMeta <- "S=" *> partyMeta A.decimal-      msgBody <- A.takeByteString-      return $ ACmd SAgent MSG {recipientMeta, brokerMeta, senderMeta, msgIntegrity, msgBody}-    replyMode = ReplyMode <$> (" NO_REPLY" $> Off <|> pure On)-    partyMeta idParser = (,) <$> idParser <* "," <*> tsISO8601P <* A.space+    msgErrResp = ACmd SAgent <$> (MERR <$> A.decimal <* A.space <*> agentErrorTypeP)+    message = ACmd SAgent <$> (MSG <$> msgMetaP <* A.space <*> A.takeByteString)+    ackCmd = ACmd SClient . ACK <$> A.decimal+    msgMetaP = do+      integrity <- msgIntegrityP+      recipient <- " R=" *> partyMeta A.decimal+      broker <- " B=" *> partyMeta base64P+      sender <- " S=" *> partyMeta A.decimal+      pure MsgMeta {integrity, recipient, broker, sender}+    partyMeta idParser = (,) <$> idParser <* "," <*> tsISO8601P     agentError = ACmd SAgent . ERR <$> agentErrorTypeP  -- | Message integrity validation result parser.@@ -478,33 +514,37 @@ serializeCommand = \case   NEW -> "NEW"   INV qInfo -> "INV " <> serializeSmpQueueInfo qInfo-  JOIN qInfo rMode -> "JOIN " <> serializeSmpQueueInfo qInfo <> replyMode rMode+  JOIN qInfo cInfo -> "JOIN " <> serializeSmpQueueInfo qInfo <> " " <> serializeBinary cInfo+  REQ confId cInfo -> "REQ " <> confId <> " " <> serializeBinary cInfo+  ACPT confId cInfo -> "ACPT " <> confId <> " " <> serializeBinary cInfo+  INFO cInfo -> "INFO " <> serializeBinary cInfo   SUB -> "SUB"-  SUBALL -> "SUBALL" -- TODO remove - hack for subscribing to all   END -> "END"-  SEND msgBody -> "SEND " <> serializeMsg msgBody+  DOWN -> "DOWN"+  UP -> "UP"+  SEND msgBody -> "SEND " <> serializeBinary msgBody+  MID mId -> "MID " <> bshow mId   SENT mId -> "SENT " <> bshow mId-  MSG {recipientMeta = (rmId, rTs), brokerMeta = (bmId, bTs), senderMeta = (smId, sTs), msgIntegrity, msgBody} ->-    B.unwords-      [ "MSG",-        serializeMsgIntegrity msgIntegrity,-        "R=" <> bshow rmId <> "," <> showTs rTs,-        "B=" <> encode bmId <> "," <> showTs bTs,-        "S=" <> bshow smId <> "," <> showTs sTs,-        serializeMsg msgBody-      ]+  MERR mId e -> "MERR " <> bshow mId <> " " <> serializeAgentError e+  MSG msgMeta msgBody ->+    "MSG " <> serializeMsgMeta msgMeta <> " " <> serializeBinary msgBody+  ACK mId -> "ACK " <> bshow mId   OFF -> "OFF"   DEL -> "DEL"   CON -> "CON"   ERR e -> "ERR " <> serializeAgentError e   OK -> "OK"   where-    replyMode :: ReplyMode -> ByteString-    replyMode = \case-      ReplyMode Off -> " NO_REPLY"-      ReplyMode On -> ""     showTs :: UTCTime -> ByteString     showTs = B.pack . formatISO8601Millis+    serializeMsgMeta :: MsgMeta -> ByteString+    serializeMsgMeta MsgMeta {integrity, recipient = (rmId, rTs), broker = (bmId, bTs), sender = (smId, sTs)} =+      B.unwords+        [ serializeMsgIntegrity integrity,+          "R=" <> bshow rmId <> "," <> showTs rTs,+          "B=" <> encode bmId <> "," <> showTs bTs,+          "S=" <> bshow smId <> "," <> showTs sTs+        ]  -- | Serialize message integrity validation result. serializeMsgIntegrity :: MsgIntegrity -> ByteString@@ -535,14 +575,19 @@   BROKER (TRANSPORT e) -> "BROKER TRANSPORT " <> serializeTransportError e   e -> bshow e -serializeMsg :: ByteString -> ByteString-serializeMsg body = bshow (B.length body) <> "\n" <> body+binaryBodyP :: Parser ByteString+binaryBodyP = do+  size :: Int <- A.decimal <* A.endOfLine+  A.take size +serializeBinary :: ByteString -> ByteString+serializeBinary body = bshow (B.length body) <> "\n" <> body+ -- | Send raw (unparsed) SMP agent protocol transmission to TCP connection. tPutRaw :: Transport c => c -> ARawTransmission -> IO ()-tPutRaw h (corrId, connAlias, command) = do+tPutRaw h (corrId, entity, command) = do   putLn h corrId-  putLn h connAlias+  putLn h entity   putLn h command  -- | Receive raw (unparsed) SMP agent protocol transmission from TCP connection.@@ -551,7 +596,7 @@  -- | Send SMP agent protocol command (or response) to TCP connection. tPut :: (Transport c, MonadIO m) => c -> ATransmission p -> m ()-tPut h (CorrId corrId, connAlias, command) =+tPut h (corrId, connAlias, command) =   liftIO $ tPutRaw h (corrId, connAlias, serializeCommand command)  -- | Receive client and agent transmissions from TCP connection.@@ -559,38 +604,42 @@ tGet party h = liftIO (tGetRaw h) >>= tParseLoadBody   where     tParseLoadBody :: ARawTransmission -> m (ATransmissionOrError p)-    tParseLoadBody t@(corrId, connAlias, command) = do-      let cmd = parseCommand command >>= fromParty >>= tConnAlias t+    tParseLoadBody t@(corrId, connId, command) = do+      let cmd = parseCommand command >>= fromParty >>= tConnId t       fullCmd <- either (return . Left) cmdWithMsgBody cmd-      return (CorrId corrId, connAlias, fullCmd)+      return (corrId, connId, fullCmd)      fromParty :: ACmd -> Either AgentErrorType (ACommand p)     fromParty (ACmd (p :: p1) cmd) = case testEquality party p of       Just Refl -> Right cmd       _ -> Left $ CMD PROHIBITED -    tConnAlias :: ARawTransmission -> ACommand p -> Either AgentErrorType (ACommand p)-    tConnAlias (_, connAlias, _) cmd = case cmd of-      -- NEW and JOIN have optional connAlias+    tConnId :: ARawTransmission -> ACommand p -> Either AgentErrorType (ACommand p)+    tConnId (_, connId, _) cmd = case cmd of+      -- NEW, JOIN and ACPT have optional connId       NEW -> Right cmd-      JOIN _ _ -> Right cmd-      -- ERROR response does not always have connAlias+      JOIN {} -> Right cmd+      -- ERROR response does not always have connId       ERR _ -> Right cmd-      -- other responses must have connAlias+      -- other responses must have connId       _-        | B.null connAlias -> Left $ CMD NO_CONN+        | B.null connId -> Left $ CMD NO_CONN         | otherwise -> Right cmd      cmdWithMsgBody :: ACommand p -> m (Either AgentErrorType (ACommand p))     cmdWithMsgBody = \case-      SEND body -> SEND <$$> getMsgBody body-      MSG agentMsgId srvTS agentTS integrity body -> MSG agentMsgId srvTS agentTS integrity <$$> getMsgBody body-      cmd -> return $ Right cmd+      SEND body -> SEND <$$> getBody body+      MSG msgMeta body -> MSG msgMeta <$$> getBody body+      JOIN qInfo cInfo -> JOIN qInfo <$$> getBody cInfo+      REQ confId cInfo -> REQ confId <$$> getBody cInfo+      ACPT confId cInfo -> ACPT confId <$$> getBody cInfo+      INFO cInfo -> INFO <$$> getBody cInfo+      cmd -> pure $ Right cmd      -- TODO refactor with server-    getMsgBody :: MsgBody -> m (Either AgentErrorType MsgBody)-    getMsgBody msgBody =-      case B.unpack msgBody of+    getBody :: ByteString -> m (Either AgentErrorType ByteString)+    getBody binary =+      case B.unpack binary of         ':' : body -> return . Right $ B.pack body         str -> case readMaybe str :: Maybe Int of           Just size -> liftIO $ do
+ src/Simplex/Messaging/Agent/RetryInterval.hs view
@@ -0,0 +1,28 @@+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Simplex.Messaging.Agent.RetryInterval where++import Control.Concurrent (threadDelay)+import Control.Monad.IO.Class (MonadIO, liftIO)++data RetryInterval = RetryInterval+  { initialInterval :: Int,+    increaseAfter :: Int,+    maxInterval :: Int+  }++withRetryInterval :: forall m. MonadIO m => RetryInterval -> (m () -> m ()) -> m ()+withRetryInterval RetryInterval {initialInterval, increaseAfter, maxInterval} action =+  callAction 0 initialInterval+  where+    callAction :: Int -> Int -> m ()+    callAction elapsedTime delay = action loop+      where+        loop = do+          let newDelay =+                if elapsedTime < increaseAfter || delay == maxInterval+                  then delay+                  else min (delay * 3 `div` 2) maxInterval+          liftIO $ threadDelay delay+          callAction (elapsedTime + delay) newDelay
src/Simplex/Messaging/Agent/Store.hs view
@@ -9,7 +9,9 @@  module Simplex.Messaging.Agent.Store where +import Control.Concurrent.STM (TVar) import Control.Exception (Exception)+import Crypto.Random (ChaChaDRG) import Data.ByteString.Char8 (ByteString) import Data.Int (Int64) import Data.Kind (Type)@@ -30,26 +32,36 @@ -- | Store class type. Defines store access methods for implementations. class Monad m => MonadAgentStore s m where   -- Queue and Connection management-  createRcvConn :: s -> RcvQueue -> m ()-  createSndConn :: s -> SndQueue -> m ()-  getConn :: s -> ConnAlias -> m SomeConn-  getAllConnAliases :: s -> m [ConnAlias] -- TODO remove - hack for subscribing to all+  createRcvConn :: s -> TVar ChaChaDRG -> ConnData -> RcvQueue -> m ConnId+  createSndConn :: s -> TVar ChaChaDRG -> ConnData -> SndQueue -> m ConnId+  getConn :: s -> ConnId -> m SomeConn+  getAllConnIds :: s -> m [ConnId] -- TODO remove - hack for subscribing to all   getRcvConn :: s -> SMPServer -> SMP.RecipientId -> m SomeConn-  deleteConn :: s -> ConnAlias -> m ()-  upgradeRcvConnToDuplex :: s -> ConnAlias -> SndQueue -> m ()-  upgradeSndConnToDuplex :: s -> ConnAlias -> RcvQueue -> m ()+  deleteConn :: s -> ConnId -> m ()+  upgradeRcvConnToDuplex :: s -> ConnId -> SndQueue -> m ()+  upgradeSndConnToDuplex :: s -> ConnId -> RcvQueue -> m ()   setRcvQueueStatus :: s -> RcvQueue -> QueueStatus -> m ()   setRcvQueueActive :: s -> RcvQueue -> VerificationKey -> m ()   setSndQueueStatus :: s -> SndQueue -> QueueStatus -> m ()--  -- Msg management-  updateRcvIds :: s -> RcvQueue -> m (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash)-  createRcvMsg :: s -> RcvQueue -> RcvMsgData -> m ()+  updateSignKey :: s -> SndQueue -> SignatureKey -> m () -  updateSndIds :: s -> SndQueue -> m (InternalId, InternalSndId, PrevSndMsgHash)-  createSndMsg :: s -> SndQueue -> SndMsgData -> m ()+  -- Confirmations+  createConfirmation :: s -> TVar ChaChaDRG -> NewConfirmation -> m ConfirmationId+  acceptConfirmation :: s -> ConfirmationId -> ConnInfo -> m AcceptedConfirmation+  getAcceptedConfirmation :: s -> ConnId -> m AcceptedConfirmation+  removeConfirmations :: s -> ConnId -> m () -  getMsg :: s -> ConnAlias -> InternalId -> m Msg+  -- Msg management+  updateRcvIds :: s -> ConnId -> m (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash)+  createRcvMsg :: s -> ConnId -> RcvMsgData -> m ()+  updateSndIds :: s -> ConnId -> m (InternalId, InternalSndId, PrevSndMsgHash)+  createSndMsg :: s -> ConnId -> SndMsgData -> m ()+  updateSndMsgStatus :: s -> ConnId -> InternalId -> SndMsgStatus -> m ()+  getPendingMsgData :: s -> ConnId -> InternalId -> m (SndQueue, MsgBody)+  getPendingMsgs :: s -> ConnId -> m [PendingMsg]+  getMsg :: s -> ConnId -> InternalId -> m Msg+  checkRcvMsg :: s -> ConnId -> InternalId -> m ()+  updateRcvMsgAck :: s -> ConnId -> InternalId -> m ()  -- * Queue types @@ -57,10 +69,8 @@ data RcvQueue = RcvQueue   { server :: SMPServer,     rcvId :: SMP.RecipientId,-    connAlias :: ConnAlias,     rcvPrivateKey :: RecipientPrivateKey,     sndId :: Maybe SMP.SenderId,-    sndKey :: Maybe SenderPublicKey,     decryptKey :: DecryptionKey,     verifyKey :: Maybe VerificationKey,     status :: QueueStatus@@ -71,7 +81,6 @@ data SndQueue = SndQueue   { server :: SMPServer,     sndId :: SMP.SenderId,-    connAlias :: ConnAlias,     sndPrivateKey :: SenderPrivateKey,     encryptKey :: EncryptionKey,     signKey :: SignatureKey,@@ -95,9 +104,9 @@ -- - DuplexConnection is a connection that has both receive and send queues set up, --   typically created by upgrading a receive or a send connection with a missing queue. data Connection (d :: ConnType) where-  RcvConnection :: ConnAlias -> RcvQueue -> Connection CRcv-  SndConnection :: ConnAlias -> SndQueue -> Connection CSnd-  DuplexConnection :: ConnAlias -> RcvQueue -> SndQueue -> Connection CDuplex+  RcvConnection :: ConnData -> RcvQueue -> Connection CRcv+  SndConnection :: ConnData -> SndQueue -> Connection CSnd+  DuplexConnection :: ConnData -> RcvQueue -> SndQueue -> Connection CDuplex  deriving instance Eq (Connection d) @@ -134,10 +143,27 @@  deriving instance Show SomeConn --- * Message integrity validation types+newtype ConnData = ConnData {connId :: ConnId}+  deriving (Eq, Show) -type MsgHash = ByteString+-- * Confirmation types +data NewConfirmation = NewConfirmation+  { connId :: ConnId,+    senderKey :: SenderPublicKey,+    senderConnInfo :: ConnInfo+  }++data AcceptedConfirmation = AcceptedConfirmation+  { confirmationId :: ConfirmationId,+    connId :: ConnId,+    senderKey :: SenderPublicKey,+    senderConnInfo :: ConnInfo,+    ownConnInfo :: ConnInfo+  }++-- * Message integrity validation types+ -- | Corresponds to `last_external_snd_msg_id` in `connections` table type PrevExternalSndId = Int64 @@ -152,15 +178,11 @@ -- * Message data containers - used on Msg creation to reduce number of parameters  data RcvMsgData = RcvMsgData-  { internalId :: InternalId,-    internalRcvId :: InternalRcvId,-    internalTs :: InternalTs,-    senderMeta :: (ExternalSndId, ExternalSndTs),-    brokerMeta :: (BrokerId, BrokerTs),+  { msgMeta :: MsgMeta,     msgBody :: MsgBody,+    internalRcvId :: InternalRcvId,     internalHash :: MsgHash,-    externalPrevSndHash :: MsgHash,-    msgIntegrity :: MsgIntegrity+    externalPrevSndHash :: MsgHash   }  data SndMsgData = SndMsgData@@ -168,9 +190,20 @@     internalSndId :: InternalSndId,     internalTs :: InternalTs,     msgBody :: MsgBody,-    internalHash :: MsgHash+    internalHash :: MsgHash,+    previousMsgHash :: MsgHash   } +data PendingMsg = PendingMsg+  { connId :: ConnId,+    msgId :: InternalId+  }+  deriving (Show)++-- * Broadcast types++type BroadcastId = ByteString+ -- * Message types  -- | A message in either direction that is stored by the agent.@@ -241,9 +274,9 @@ newtype InternalSndId = InternalSndId {unSndId :: Int64} deriving (Eq, Show)  data SndMsgStatus-  = Created-  | Sent-  | Delivered+  = SndMsgCreated+  | SndMsgSent+  | SndMsgDelivered   deriving (Eq, Show)  type SentTs = UTCTime@@ -252,7 +285,7 @@  -- | Base message data independent of direction. data MsgBase = MsgBase-  { connAlias :: ConnAlias,+  { connAlias :: ConnId,     -- | Monotonically increasing id of a message per connection, internal to the agent.     -- Internal Id preserves ordering between both received and sent messages, and is needed     -- to track the order of the conversation (which can be different for the sender / receiver)@@ -276,6 +309,8 @@ data StoreError   = -- | IO exceptions in store actions.     SEInternal ByteString+  | -- | failed to generate unique random ID+    SEUniqueID   | -- | Connection alias not found (or both queues absent).     SEConnNotFound   | -- | Connection alias already used.@@ -283,6 +318,10 @@   | -- | Wrong connection type, e.g. "send" connection when "receive" or "duplex" is expected, or vice versa.     -- 'upgradeRcvConnToDuplex' and 'upgradeSndConnToDuplex' do not allow duplex connections - they would also return this error.     SEBadConnType ConnType+  | -- | Confirmation not found.+    SEConfirmationNotFound+  | -- | Message not found+    SEMsgNotFound   | -- | Currently not used. The intention was to pass current expected queue status in methods,     -- as we always know what it should be at any stage of the protocol,     -- and in case it does not match use this error.
src/Simplex/Messaging/Agent/Store/SQLite.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE DataKinds #-} {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-}@@ -11,6 +12,7 @@ {-# LANGUAGE QuasiQuotes #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TupleSections #-} {-# LANGUAGE UndecidableInstances #-} {-# OPTIONS_GHC -fno-warn-orphans #-} @@ -18,19 +20,25 @@   ( SQLiteStore (..),     createSQLiteStore,     connectSQLiteStore,+    withConnection,+    withTransaction,+    fromTextField_,   ) where  import Control.Concurrent (threadDelay)-import Control.Monad (when)-import Control.Monad.Except (MonadError (throwError), MonadIO (liftIO))+import Control.Concurrent.STM+import Control.Exception (bracket)+import Control.Monad.Except import Control.Monad.IO.Unlift (MonadUnliftIO)-import Data.Bifunctor (first)+import Crypto.Random (ChaChaDRG, randomBytesGenerate)+import Data.ByteString (ByteString)+import Data.ByteString.Base64 (encode)+import Data.Char (toLower) import Data.List (find) import Data.Maybe (fromMaybe)-import Data.Text (isPrefixOf)+import Data.Text (Text) import qualified Data.Text as T-import Data.Text.Encoding (decodeUtf8) import Database.SQLite.Simple (FromRow, NamedParam (..), Only (..), SQLData (..), SQLError, field) import qualified Database.SQLite.Simple as DB import Database.SQLite.Simple.FromField@@ -41,105 +49,150 @@ import Network.Socket (ServiceName) import Simplex.Messaging.Agent.Protocol import Simplex.Messaging.Agent.Store-import Simplex.Messaging.Agent.Store.SQLite.Schema (createSchema)+import Simplex.Messaging.Agent.Store.SQLite.Migrations (Migration)+import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations import Simplex.Messaging.Parsers (blobFieldParser)+import Simplex.Messaging.Protocol (MsgBody) import qualified Simplex.Messaging.Protocol as SMP import Simplex.Messaging.Util (bshow, liftIOEither)-import System.Exit (ExitCode (ExitFailure), exitWith)+import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist)+import System.Exit (exitFailure) import System.FilePath (takeDirectory)+import System.IO (hFlush, stdout) import Text.Read (readMaybe)-import UnliftIO.Directory (createDirectoryIfMissing) import qualified UnliftIO.Exception as E  -- * SQLite Store implementation  data SQLiteStore = SQLiteStore   { dbFilePath :: FilePath,-    dbConn :: DB.Connection+    dbConnPool :: TBQueue DB.Connection,+    dbNew :: Bool   } -createSQLiteStore :: MonadUnliftIO m => FilePath -> m SQLiteStore-createSQLiteStore dbFilePath = do+createSQLiteStore :: FilePath -> Int -> [Migration] -> IO SQLiteStore+createSQLiteStore dbFilePath poolSize migrations = do   let dbDir = takeDirectory dbFilePath   createDirectoryIfMissing False dbDir-  store <- connectSQLiteStore dbFilePath-  compileOptions <- liftIO (DB.query_ (dbConn store) "pragma COMPILE_OPTIONS;" :: IO [[T.Text]])-  let threadsafeOption = find (isPrefixOf "THREADSAFE=") (concat compileOptions)-  liftIO $ case threadsafeOption of-    Just "THREADSAFE=0" -> do-      putStrLn "SQLite compiled with not threadsafe code, continue (y/n):"-      s <- getLine-      when (s /= "y") (exitWith $ ExitFailure 2)+  st <- connectSQLiteStore dbFilePath poolSize+  checkThreadsafe st+  migrateSchema st migrations+  pure st++checkThreadsafe :: SQLiteStore -> IO ()+checkThreadsafe st = withConnection st $ \db -> do+  compileOptions <- DB.query_ db "pragma COMPILE_OPTIONS;" :: IO [[Text]]+  let threadsafeOption = find (T.isPrefixOf "THREADSAFE=") (concat compileOptions)+  case threadsafeOption of+    Just "THREADSAFE=0" -> confirmOrExit "SQLite compiled with non-threadsafe code."     Nothing -> putStrLn "Warning: SQLite THREADSAFE compile option not found"     _ -> return ()-  liftIO . createSchema $ dbConn store-  return store -connectSQLiteStore :: MonadUnliftIO m => FilePath -> m SQLiteStore-connectSQLiteStore dbFilePath = do-  dbConn <- liftIO $ DB.open dbFilePath-  liftIO $-    DB.execute_-      dbConn-      [sql|-        PRAGMA foreign_keys = ON;-        PRAGMA journal_mode = WAL;-      |]-  return SQLiteStore {dbFilePath, dbConn}+migrateSchema :: SQLiteStore -> [Migration] -> IO ()+migrateSchema st migrations = withConnection st $ \db -> do+  Migrations.initialize db+  Migrations.get db migrations >>= \case+    Left e -> confirmOrExit $ "Database error: " <> e+    Right [] -> pure ()+    Right ms -> do+      unless (dbNew st) $ do+        confirmOrExit "The app has a newer version than the database - it will be backed up and upgraded."+        let f = dbFilePath st+        copyFile f (f <> ".bak")+      Migrations.run db ms -checkDuplicate :: (MonadUnliftIO m, MonadError StoreError m) => IO () -> m ()-checkDuplicate action = liftIOEither $ first handleError <$> E.try action-  where-    handleError :: SQLError -> StoreError-    handleError e-      | DB.sqlError e == DB.ErrorConstraint = SEConnDuplicate-      | otherwise = SEInternal $ bshow e+confirmOrExit :: String -> IO ()+confirmOrExit s = do+  putStrLn s+  putStr "Continue (y/N): "+  hFlush stdout+  ok <- getLine+  when (map toLower ok /= "y") exitFailure -withTransaction :: forall a. DB.Connection -> IO a -> IO a-withTransaction db a = loop 100 100_000+connectSQLiteStore :: FilePath -> Int -> IO SQLiteStore+connectSQLiteStore dbFilePath poolSize = do+  dbNew <- not <$> doesFileExist dbFilePath+  dbConnPool <- newTBQueueIO $ toEnum poolSize+  replicateM_ poolSize $+    connectDB dbFilePath >>= atomically . writeTBQueue dbConnPool+  pure SQLiteStore {dbFilePath, dbConnPool, dbNew}++connectDB :: FilePath -> IO DB.Connection+connectDB path = do+  dbConn <- DB.open path+  DB.execute_ dbConn "PRAGMA foreign_keys = ON; PRAGMA journal_mode = WAL;"+  pure dbConn++checkConstraint :: StoreError -> IO (Either StoreError a) -> IO (Either StoreError a)+checkConstraint err action = action `E.catch` (pure . Left . handleSQLError err)++handleSQLError :: StoreError -> SQLError -> StoreError+handleSQLError err e+  | DB.sqlError e == DB.ErrorConstraint = err+  | otherwise = SEInternal $ bshow e++withConnection :: SQLiteStore -> (DB.Connection -> IO a) -> IO a+withConnection SQLiteStore {dbConnPool} =+  bracket+    (atomically $ readTBQueue dbConnPool)+    (atomically . writeTBQueue dbConnPool)++withTransaction :: forall a. SQLiteStore -> (DB.Connection -> IO a) -> IO a+withTransaction st action = withConnection st $ loop 100 100_000   where-    loop :: Int -> Int -> IO a-    loop t tLim =-      DB.withImmediateTransaction db a `E.catch` \(e :: SQLError) ->+    loop :: Int -> Int -> DB.Connection -> IO a+    loop t tLim db =+      DB.withImmediateTransaction db (action db) `E.catch` \(e :: SQLError) ->         if tLim > t && DB.sqlError e == DB.ErrorBusy           then do             threadDelay t-            loop (t * 9 `div` 8) (tLim - t)+            loop (t * 9 `div` 8) (tLim - t) db           else E.throwIO e  instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteStore m where-  createRcvConn :: SQLiteStore -> RcvQueue -> m ()-  createRcvConn SQLiteStore {dbConn} q@RcvQueue {server} =-    checkDuplicate $-      withTransaction dbConn $ do-        upsertServer_ dbConn server-        insertRcvQueue_ dbConn q-        insertRcvConnection_ dbConn q+  createRcvConn :: SQLiteStore -> TVar ChaChaDRG -> ConnData -> RcvQueue -> m ConnId+  createRcvConn st gVar cData q@RcvQueue {server} =+    -- TODO if schema has to be restarted, this function can be refactored+    -- to create connection first using createWithRandomId+    liftIOEither . checkConstraint SEConnDuplicate . withTransaction st $ \db ->+      getConnId_ db gVar cData >>= traverse (create db)+    where+      create :: DB.Connection -> ConnId -> IO ConnId+      create db connId = do+        upsertServer_ db server+        insertRcvQueue_ db connId q+        insertRcvConnection_ db cData {connId} q+        pure connId -  createSndConn :: SQLiteStore -> SndQueue -> m ()-  createSndConn SQLiteStore {dbConn} q@SndQueue {server} =-    checkDuplicate $-      withTransaction dbConn $ do-        upsertServer_ dbConn server-        insertSndQueue_ dbConn q-        insertSndConnection_ dbConn q+  createSndConn :: SQLiteStore -> TVar ChaChaDRG -> ConnData -> SndQueue -> m ConnId+  createSndConn st gVar cData q@SndQueue {server} =+    -- TODO if schema has to be restarted, this function can be refactored+    -- to create connection first using createWithRandomId+    liftIOEither . checkConstraint SEConnDuplicate . withTransaction st $ \db ->+      getConnId_ db gVar cData >>= traverse (create db)+    where+      create :: DB.Connection -> ConnId -> IO ConnId+      create db connId = do+        upsertServer_ db server+        insertSndQueue_ db connId q+        insertSndConnection_ db cData {connId} q+        pure connId -  getConn :: SQLiteStore -> ConnAlias -> m SomeConn-  getConn SQLiteStore {dbConn} connAlias =-    liftIOEither . withTransaction dbConn $-      getConn_ dbConn connAlias+  getConn :: SQLiteStore -> ConnId -> m SomeConn+  getConn st connId =+    liftIOEither . withTransaction st $ \db ->+      getConn_ db connId -  getAllConnAliases :: SQLiteStore -> m [ConnAlias]-  getAllConnAliases SQLiteStore {dbConn} =-    liftIO $ do-      r <- DB.query_ dbConn "SELECT conn_alias FROM connections;" :: IO [[ConnAlias]]-      return (concat r)+  getAllConnIds :: SQLiteStore -> m [ConnId]+  getAllConnIds st =+    liftIO . withTransaction st $ \db ->+      concat <$> (DB.query_ db "SELECT conn_alias FROM connections;" :: IO [[ConnId]])    getRcvConn :: SQLiteStore -> SMPServer -> SMP.RecipientId -> m SomeConn-  getRcvConn SQLiteStore {dbConn} SMPServer {host, port} rcvId =-    liftIOEither . withTransaction dbConn $+  getRcvConn st SMPServer {host, port} rcvId =+    liftIOEither . withTransaction st $ \db ->       DB.queryNamed-        dbConn+        db         [sql|           SELECT q.conn_alias           FROM rcv_queues q@@ -147,47 +200,47 @@         |]         [":host" := host, ":port" := serializePort_ port, ":rcv_id" := rcvId]         >>= \case-          [Only connAlias] -> getConn_ dbConn connAlias+          [Only connId] -> getConn_ db connId           _ -> pure $ Left SEConnNotFound -  deleteConn :: SQLiteStore -> ConnAlias -> m ()-  deleteConn SQLiteStore {dbConn} connAlias =-    liftIO $+  deleteConn :: SQLiteStore -> ConnId -> m ()+  deleteConn st connId =+    liftIO . withTransaction st $ \db ->       DB.executeNamed-        dbConn+        db         "DELETE FROM connections WHERE conn_alias = :conn_alias;"-        [":conn_alias" := connAlias]+        [":conn_alias" := connId] -  upgradeRcvConnToDuplex :: SQLiteStore -> ConnAlias -> SndQueue -> m ()-  upgradeRcvConnToDuplex SQLiteStore {dbConn} connAlias sq@SndQueue {server} =-    liftIOEither . withTransaction dbConn $-      getConn_ dbConn connAlias >>= \case-        Right (SomeConn SCRcv (RcvConnection _ _)) -> do-          upsertServer_ dbConn server-          insertSndQueue_ dbConn sq-          updateConnWithSndQueue_ dbConn connAlias sq+  upgradeRcvConnToDuplex :: SQLiteStore -> ConnId -> SndQueue -> m ()+  upgradeRcvConnToDuplex st connId sq@SndQueue {server} =+    liftIOEither . withTransaction st $ \db ->+      getConn_ db connId >>= \case+        Right (SomeConn _ RcvConnection {}) -> do+          upsertServer_ db server+          insertSndQueue_ db connId sq+          updateConnWithSndQueue_ db connId sq           pure $ Right ()         Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c         _ -> pure $ Left SEConnNotFound -  upgradeSndConnToDuplex :: SQLiteStore -> ConnAlias -> RcvQueue -> m ()-  upgradeSndConnToDuplex SQLiteStore {dbConn} connAlias rq@RcvQueue {server} =-    liftIOEither . withTransaction dbConn $-      getConn_ dbConn connAlias >>= \case-        Right (SomeConn SCSnd (SndConnection _ _)) -> do-          upsertServer_ dbConn server-          insertRcvQueue_ dbConn rq-          updateConnWithRcvQueue_ dbConn connAlias rq+  upgradeSndConnToDuplex :: SQLiteStore -> ConnId -> RcvQueue -> m ()+  upgradeSndConnToDuplex st connId rq@RcvQueue {server} =+    liftIOEither . withTransaction st $ \db ->+      getConn_ db connId >>= \case+        Right (SomeConn _ SndConnection {}) -> do+          upsertServer_ db server+          insertRcvQueue_ db connId rq+          updateConnWithRcvQueue_ db connId rq           pure $ Right ()         Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c         _ -> pure $ Left SEConnNotFound    setRcvQueueStatus :: SQLiteStore -> RcvQueue -> QueueStatus -> m ()-  setRcvQueueStatus SQLiteStore {dbConn} RcvQueue {rcvId, server = SMPServer {host, port}} status =-    -- ? throw error if queue doesn't exist?-    liftIO $+  setRcvQueueStatus st RcvQueue {rcvId, server = SMPServer {host, port}} status =+    -- ? throw error if queue does not exist?+    liftIO . withTransaction st $ \db ->       DB.executeNamed-        dbConn+        db         [sql|           UPDATE rcv_queues           SET status = :status@@ -196,11 +249,11 @@         [":status" := status, ":host" := host, ":port" := serializePort_ port, ":rcv_id" := rcvId]    setRcvQueueActive :: SQLiteStore -> RcvQueue -> VerificationKey -> m ()-  setRcvQueueActive SQLiteStore {dbConn} RcvQueue {rcvId, server = SMPServer {host, port}} verifyKey =-    -- ? throw error if queue doesn't exist?-    liftIO $+  setRcvQueueActive st RcvQueue {rcvId, server = SMPServer {host, port}} verifyKey =+    -- ? throw error if queue does not exist?+    liftIO . withTransaction st $ \db ->       DB.executeNamed-        dbConn+        db         [sql|           UPDATE rcv_queues           SET verify_key = :verify_key, status = :status@@ -214,11 +267,11 @@         ]    setSndQueueStatus :: SQLiteStore -> SndQueue -> QueueStatus -> m ()-  setSndQueueStatus SQLiteStore {dbConn} SndQueue {sndId, server = SMPServer {host, port}} status =-    -- ? throw error if queue doesn't exist?-    liftIO $+  setSndQueueStatus st SndQueue {sndId, server = SMPServer {host, port}} status =+    -- ? throw error if queue does not exist?+    liftIO . withTransaction st $ \db ->       DB.executeNamed-        dbConn+        db         [sql|           UPDATE snd_queues           SET status = :status@@ -226,41 +279,194 @@         |]         [":status" := status, ":host" := host, ":port" := serializePort_ port, ":snd_id" := sndId] -  updateRcvIds :: SQLiteStore -> RcvQueue -> m (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash)-  updateRcvIds SQLiteStore {dbConn} RcvQueue {connAlias} =-    liftIO . withTransaction dbConn $ do-      (lastInternalId, lastInternalRcvId, lastExternalSndId, lastRcvHash) <- retrieveLastIdsAndHashRcv_ dbConn connAlias+  updateSignKey :: SQLiteStore -> SndQueue -> SignatureKey -> m ()+  updateSignKey st SndQueue {sndId, server = SMPServer {host, port}} signatureKey =+    liftIO . withTransaction st $ \db ->+      DB.executeNamed+        db+        [sql|+          UPDATE snd_queues+          SET sign_key = :sign_key+          WHERE host = :host AND port = :port AND snd_id = :snd_id;+        |]+        [":sign_key" := signatureKey, ":host" := host, ":port" := serializePort_ port, ":snd_id" := sndId]++  createConfirmation :: SQLiteStore -> TVar ChaChaDRG -> NewConfirmation -> m ConfirmationId+  createConfirmation st gVar NewConfirmation {connId, senderKey, senderConnInfo} =+    liftIOEither . withTransaction st $ \db ->+      createWithRandomId gVar $ \confirmationId ->+        DB.execute+          db+          [sql|+            INSERT INTO conn_confirmations+            (confirmation_id, conn_alias, sender_key, sender_conn_info, accepted) VALUES (?, ?, ?, ?, 0);+          |]+          (confirmationId, connId, senderKey, senderConnInfo)++  acceptConfirmation :: SQLiteStore -> ConfirmationId -> ConnInfo -> m AcceptedConfirmation+  acceptConfirmation st confirmationId ownConnInfo =+    liftIOEither . withTransaction st $ \db -> do+      DB.executeNamed+        db+        [sql|+          UPDATE conn_confirmations+          SET accepted = 1,+              own_conn_info = :own_conn_info+          WHERE confirmation_id = :confirmation_id;+        |]+        [ ":own_conn_info" := ownConnInfo,+          ":confirmation_id" := confirmationId+        ]+      confirmation+        <$> DB.query+          db+          [sql|+            SELECT conn_alias, sender_key, sender_conn_info+            FROM conn_confirmations+            WHERE confirmation_id = ?;+          |]+          (Only confirmationId)+    where+      confirmation [(connId, senderKey, senderConnInfo)] =+        Right $ AcceptedConfirmation {confirmationId, connId, senderKey, senderConnInfo, ownConnInfo}+      confirmation _ = Left SEConfirmationNotFound++  getAcceptedConfirmation :: SQLiteStore -> ConnId -> m AcceptedConfirmation+  getAcceptedConfirmation st connId =+    liftIOEither . withTransaction st $ \db ->+      confirmation+        <$> DB.query+          db+          [sql|+            SELECT confirmation_id, sender_key, sender_conn_info, own_conn_info+            FROM conn_confirmations+            WHERE conn_alias = ? AND accepted = 1;+          |]+          (Only connId)+    where+      confirmation [(confirmationId, senderKey, senderConnInfo, ownConnInfo)] =+        Right $ AcceptedConfirmation {confirmationId, connId, senderKey, senderConnInfo, ownConnInfo}+      confirmation _ = Left SEConfirmationNotFound++  removeConfirmations :: SQLiteStore -> ConnId -> m ()+  removeConfirmations st connId =+    liftIO . withTransaction st $ \db ->+      DB.executeNamed+        db+        [sql|+          DELETE FROM conn_confirmations+          WHERE conn_alias = :conn_alias;+        |]+        [":conn_alias" := connId]++  updateRcvIds :: SQLiteStore -> ConnId -> m (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash)+  updateRcvIds st connId =+    liftIO . withTransaction st $ \db -> do+      (lastInternalId, lastInternalRcvId, lastExternalSndId, lastRcvHash) <- retrieveLastIdsAndHashRcv_ db connId       let internalId = InternalId $ unId lastInternalId + 1           internalRcvId = InternalRcvId $ unRcvId lastInternalRcvId + 1-      updateLastIdsRcv_ dbConn connAlias internalId internalRcvId+      updateLastIdsRcv_ db connId internalId internalRcvId       pure (internalId, internalRcvId, lastExternalSndId, lastRcvHash) -  createRcvMsg :: SQLiteStore -> RcvQueue -> RcvMsgData -> m ()-  createRcvMsg SQLiteStore {dbConn} RcvQueue {connAlias} rcvMsgData =-    liftIO . withTransaction dbConn $ do-      insertRcvMsgBase_ dbConn connAlias rcvMsgData-      insertRcvMsgDetails_ dbConn connAlias rcvMsgData-      updateHashRcv_ dbConn connAlias rcvMsgData+  createRcvMsg :: SQLiteStore -> ConnId -> RcvMsgData -> m ()+  createRcvMsg st connId rcvMsgData =+    liftIO . withTransaction st $ \db -> do+      insertRcvMsgBase_ db connId rcvMsgData+      insertRcvMsgDetails_ db connId rcvMsgData+      updateHashRcv_ db connId rcvMsgData -  updateSndIds :: SQLiteStore -> SndQueue -> m (InternalId, InternalSndId, PrevSndMsgHash)-  updateSndIds SQLiteStore {dbConn} SndQueue {connAlias} =-    liftIO . withTransaction dbConn $ do-      (lastInternalId, lastInternalSndId, prevSndHash) <- retrieveLastIdsAndHashSnd_ dbConn connAlias+  updateSndIds :: SQLiteStore -> ConnId -> m (InternalId, InternalSndId, PrevSndMsgHash)+  updateSndIds st connId =+    liftIO . withTransaction st $ \db -> do+      (lastInternalId, lastInternalSndId, prevSndHash) <- retrieveLastIdsAndHashSnd_ db connId       let internalId = InternalId $ unId lastInternalId + 1           internalSndId = InternalSndId $ unSndId lastInternalSndId + 1-      updateLastIdsSnd_ dbConn connAlias internalId internalSndId+      updateLastIdsSnd_ db connId internalId internalSndId       pure (internalId, internalSndId, prevSndHash) -  createSndMsg :: SQLiteStore -> SndQueue -> SndMsgData -> m ()-  createSndMsg SQLiteStore {dbConn} SndQueue {connAlias} sndMsgData =-    liftIO . withTransaction dbConn $ do-      insertSndMsgBase_ dbConn connAlias sndMsgData-      insertSndMsgDetails_ dbConn connAlias sndMsgData-      updateHashSnd_ dbConn connAlias sndMsgData+  createSndMsg :: SQLiteStore -> ConnId -> SndMsgData -> m ()+  createSndMsg st connId sndMsgData =+    liftIO . withTransaction st $ \db -> do+      insertSndMsgBase_ db connId sndMsgData+      insertSndMsgDetails_ db connId sndMsgData+      updateHashSnd_ db connId sndMsgData -  getMsg :: SQLiteStore -> ConnAlias -> InternalId -> m Msg-  getMsg _st _connAlias _id = throwError SENotImplemented+  updateSndMsgStatus :: SQLiteStore -> ConnId -> InternalId -> SndMsgStatus -> m ()+  updateSndMsgStatus st connId msgId msgStatus =+    liftIO . withTransaction st $ \db ->+      DB.executeNamed+        db+        [sql|+          UPDATE snd_messages+          SET snd_status = :snd_status+          WHERE conn_alias = :conn_alias AND internal_id = :internal_id+        |]+        [ ":conn_alias" := connId,+          ":internal_id" := msgId,+          ":snd_status" := msgStatus+        ] +  getPendingMsgData :: SQLiteStore -> ConnId -> InternalId -> m (SndQueue, MsgBody)+  getPendingMsgData st connId msgId =+    liftIOEither . withTransaction st $ \db -> runExceptT $ do+      sq <- ExceptT $ sndQueue <$> getSndQueueByConnAlias_ db connId+      msgBody <-+        ExceptT $+          sndMsgData+            <$> DB.query+              db+              [sql|+                SELECT m.msg_body+                FROM messages m+                JOIN snd_messages s ON s.conn_alias = m.conn_alias AND s.internal_id = m.internal_id+                WHERE m.conn_alias = ? AND m.internal_id = ?+              |]+              (connId, msgId)+      pure (sq, msgBody)+    where+      sndMsgData :: [Only MsgBody] -> Either StoreError MsgBody+      sndMsgData [Only msgBody] = Right msgBody+      sndMsgData _ = Left SEMsgNotFound+      sndQueue :: Maybe SndQueue -> Either StoreError SndQueue+      sndQueue = maybe (Left SEConnNotFound) Right++  getPendingMsgs :: SQLiteStore -> ConnId -> m [PendingMsg]+  getPendingMsgs st connId =+    liftIO . withTransaction st $ \db ->+      map (PendingMsg connId . fromOnly)+        <$> DB.query db "SELECT internal_id FROM snd_messages WHERE conn_alias = ? AND snd_status = ?" (connId, SndMsgCreated)++  getMsg :: SQLiteStore -> ConnId -> InternalId -> m Msg+  getMsg _st _connId _id = throwError SENotImplemented++  checkRcvMsg :: SQLiteStore -> ConnId -> InternalId -> m ()+  checkRcvMsg st connId msgId =+    liftIOEither . withTransaction st $ \db ->+      hasMsg+        <$> DB.query+          db+          [sql|+            SELECT conn_alias, internal_id+            FROM rcv_messages+            WHERE conn_alias = ? AND internal_id = ?+          |]+          (connId, msgId)+    where+      hasMsg :: [(ConnId, InternalId)] -> Either StoreError ()+      hasMsg r = if null r then Left SEMsgNotFound else Right ()++  updateRcvMsgAck :: SQLiteStore -> ConnId -> InternalId -> m ()+  updateRcvMsgAck st connId msgId =+    liftIO . withTransaction st $ \db -> do+      DB.execute+        db+        [sql|+          UPDATE rcv_messages+          SET rcv_status = ?, ack_brocker_ts = datetime('now')+          WHERE conn_alias = ? AND internal_id = ?+        |]+        (AcknowledgedToBroker, connId, msgId)+ -- * Auxiliary helpers  -- ? replace with ToField? - it's easy to forget to use this@@ -273,7 +479,7 @@  instance ToField QueueStatus where toField = toField . show -instance FromField QueueStatus where fromField = fromFieldToReadable_+instance FromField QueueStatus where fromField = fromTextField_ $ readMaybe . T.unpack  instance ToField InternalRcvId where toField (InternalRcvId x) = toField x @@ -295,13 +501,16 @@  instance FromField MsgIntegrity where fromField = blobFieldParser msgIntegrityP -fromFieldToReadable_ :: forall a. (Read a, E.Typeable a) => Field -> Ok a-fromFieldToReadable_ = \case+instance ToField SMPQueueInfo where toField = toField . serializeSmpQueueInfo++instance FromField SMPQueueInfo where fromField = blobFieldParser smpQueueInfoP++fromTextField_ :: (E.Typeable a) => (Text -> Maybe a) -> Field -> Ok a+fromTextField_ fromText = \case   f@(Field (SQLText t) _) ->-    let str = T.unpack t-     in case readMaybe str of-          Just x -> Ok x-          _ -> returnError ConversionFailed f ("invalid string: " <> str)+    case fromText t of+      Just x -> Ok x+      _ -> returnError ConversionFailed f ("invalid text: " <> T.unpack t)   f -> returnError ConversionFailed f "expecting SQLText column type"  {- ORMOLU_DISABLE -}@@ -333,49 +542,49 @@  -- * createRcvConn helpers -insertRcvQueue_ :: DB.Connection -> RcvQueue -> IO ()-insertRcvQueue_ dbConn RcvQueue {..} = do+insertRcvQueue_ :: DB.Connection -> ConnId -> RcvQueue -> IO ()+insertRcvQueue_ dbConn connId RcvQueue {..} = do   let port_ = serializePort_ $ port server   DB.executeNamed     dbConn     [sql|       INSERT INTO rcv_queues-        ( host, port, rcv_id, conn_alias, rcv_private_key, snd_id, snd_key, decrypt_key, verify_key, status)+        ( host, port, rcv_id, conn_alias, rcv_private_key, snd_id, decrypt_key, verify_key, status)       VALUES-        (:host,:port,:rcv_id,:conn_alias,:rcv_private_key,:snd_id,:snd_key,:decrypt_key,:verify_key,:status);+        (:host,:port,:rcv_id,:conn_alias,:rcv_private_key,:snd_id,:decrypt_key,:verify_key,:status);     |]     [ ":host" := host server,       ":port" := port_,       ":rcv_id" := rcvId,-      ":conn_alias" := connAlias,+      ":conn_alias" := connId,       ":rcv_private_key" := rcvPrivateKey,       ":snd_id" := sndId,-      ":snd_key" := sndKey,       ":decrypt_key" := decryptKey,       ":verify_key" := verifyKey,       ":status" := status     ] -insertRcvConnection_ :: DB.Connection -> RcvQueue -> IO ()-insertRcvConnection_ dbConn RcvQueue {server, rcvId, connAlias} = do+insertRcvConnection_ :: DB.Connection -> ConnData -> RcvQueue -> IO ()+insertRcvConnection_ dbConn ConnData {connId} RcvQueue {server, rcvId} = do   let port_ = serializePort_ $ port server   DB.executeNamed     dbConn     [sql|       INSERT INTO connections-        ( conn_alias, rcv_host, rcv_port, rcv_id, snd_host, snd_port, snd_id,-          last_internal_msg_id, last_internal_rcv_msg_id, last_internal_snd_msg_id,-          last_external_snd_msg_id, last_rcv_msg_hash, last_snd_msg_hash)+        ( conn_alias, rcv_host, rcv_port, rcv_id, snd_host, snd_port, snd_id, last_internal_msg_id, last_internal_rcv_msg_id, last_internal_snd_msg_id, last_external_snd_msg_id, last_rcv_msg_hash, last_snd_msg_hash)       VALUES-        (:conn_alias,:rcv_host,:rcv_port,:rcv_id,     NULL,     NULL,   NULL,-          0, 0, 0, 0, x'', x'');+        (:conn_alias,:rcv_host,:rcv_port,:rcv_id, NULL,     NULL,     NULL, 0, 0, 0, 0, x'', x'');     |]-    [":conn_alias" := connAlias, ":rcv_host" := host server, ":rcv_port" := port_, ":rcv_id" := rcvId]+    [ ":conn_alias" := connId,+      ":rcv_host" := host server,+      ":rcv_port" := port_,+      ":rcv_id" := rcvId+    ]  -- * createSndConn helpers -insertSndQueue_ :: DB.Connection -> SndQueue -> IO ()-insertSndQueue_ dbConn SndQueue {..} = do+insertSndQueue_ :: DB.Connection -> ConnId -> SndQueue -> IO ()+insertSndQueue_ dbConn connId SndQueue {..} = do   let port_ = serializePort_ $ port server   DB.executeNamed     dbConn@@ -388,85 +597,94 @@     [ ":host" := host server,       ":port" := port_,       ":snd_id" := sndId,-      ":conn_alias" := connAlias,+      ":conn_alias" := connId,       ":snd_private_key" := sndPrivateKey,       ":encrypt_key" := encryptKey,       ":sign_key" := signKey,       ":status" := status     ] -insertSndConnection_ :: DB.Connection -> SndQueue -> IO ()-insertSndConnection_ dbConn SndQueue {server, sndId, connAlias} = do+insertSndConnection_ :: DB.Connection -> ConnData -> SndQueue -> IO ()+insertSndConnection_ dbConn ConnData {connId} SndQueue {server, sndId} = do   let port_ = serializePort_ $ port server   DB.executeNamed     dbConn     [sql|       INSERT INTO connections-        ( conn_alias, rcv_host, rcv_port, rcv_id, snd_host, snd_port, snd_id,-          last_internal_msg_id, last_internal_rcv_msg_id, last_internal_snd_msg_id,-          last_external_snd_msg_id, last_rcv_msg_hash, last_snd_msg_hash)+        ( conn_alias, rcv_host, rcv_port, rcv_id, snd_host, snd_port, snd_id, last_internal_msg_id, last_internal_rcv_msg_id, last_internal_snd_msg_id, last_external_snd_msg_id, last_rcv_msg_hash, last_snd_msg_hash)       VALUES-        (:conn_alias,     NULL,     NULL,   NULL,:snd_host,:snd_port,:snd_id,-          0, 0, 0, 0, x'', x'');+        (:conn_alias, NULL,     NULL,     NULL,  :snd_host,:snd_port,:snd_id, 0, 0, 0, 0, x'', x'');     |]-    [":conn_alias" := connAlias, ":snd_host" := host server, ":snd_port" := port_, ":snd_id" := sndId]+    [ ":conn_alias" := connId,+      ":snd_host" := host server,+      ":snd_port" := port_,+      ":snd_id" := sndId+    ]  -- * getConn helpers -getConn_ :: DB.Connection -> ConnAlias -> IO (Either StoreError SomeConn)-getConn_ dbConn connAlias = do-  rQ <- retrieveRcvQueueByConnAlias_ dbConn connAlias-  sQ <- retrieveSndQueueByConnAlias_ dbConn connAlias-  pure $ case (rQ, sQ) of-    (Just rcvQ, Just sndQ) -> Right $ SomeConn SCDuplex (DuplexConnection connAlias rcvQ sndQ)-    (Just rcvQ, Nothing) -> Right $ SomeConn SCRcv (RcvConnection connAlias rcvQ)-    (Nothing, Just sndQ) -> Right $ SomeConn SCSnd (SndConnection connAlias sndQ)-    _ -> Left SEConnNotFound+getConn_ :: DB.Connection -> ConnId -> IO (Either StoreError SomeConn)+getConn_ dbConn connId =+  getConnData_ dbConn connId >>= \case+    Nothing -> pure $ Left SEConnNotFound+    Just connData -> do+      rQ <- getRcvQueueByConnAlias_ dbConn connId+      sQ <- getSndQueueByConnAlias_ dbConn connId+      pure $ case (rQ, sQ) of+        (Just rcvQ, Just sndQ) -> Right $ SomeConn SCDuplex (DuplexConnection connData rcvQ sndQ)+        (Just rcvQ, Nothing) -> Right $ SomeConn SCRcv (RcvConnection connData rcvQ)+        (Nothing, Just sndQ) -> Right $ SomeConn SCSnd (SndConnection connData sndQ)+        _ -> Left SEConnNotFound -retrieveRcvQueueByConnAlias_ :: DB.Connection -> ConnAlias -> IO (Maybe RcvQueue)-retrieveRcvQueueByConnAlias_ dbConn connAlias = do-  r <--    DB.queryNamed+getConnData_ :: DB.Connection -> ConnId -> IO (Maybe ConnData)+getConnData_ dbConn connId' =+  connData+    <$> DB.query dbConn "SELECT conn_alias FROM connections WHERE conn_alias = ?;" (Only connId')+  where+    connData [Only connId] = Just ConnData {connId}+    connData _ = Nothing++getRcvQueueByConnAlias_ :: DB.Connection -> ConnId -> IO (Maybe RcvQueue)+getRcvQueueByConnAlias_ dbConn connId =+  rcvQueue+    <$> DB.query       dbConn       [sql|-        SELECT-          s.key_hash, q.host, q.port, q.rcv_id, q.conn_alias, q.rcv_private_key,-          q.snd_id, q.snd_key, q.decrypt_key, q.verify_key, q.status+        SELECT s.key_hash, q.host, q.port, q.rcv_id, q.rcv_private_key,+          q.snd_id, q.decrypt_key, q.verify_key, q.status         FROM rcv_queues q         INNER JOIN servers s ON q.host = s.host AND q.port = s.port-        WHERE q.conn_alias = :conn_alias;+        WHERE q.conn_alias = ?;       |]-      [":conn_alias" := connAlias]-  case r of-    [(keyHash, host, port, rcvId, cAlias, rcvPrivateKey, sndId, sndKey, decryptKey, verifyKey, status)] -> do+      (Only connId)+  where+    rcvQueue [(keyHash, host, port, rcvId, rcvPrivateKey, sndId, decryptKey, verifyKey, status)] =       let srv = SMPServer host (deserializePort_ port) keyHash-      return . Just $ RcvQueue srv rcvId cAlias rcvPrivateKey sndId sndKey decryptKey verifyKey status-    _ -> return Nothing+       in Just $ RcvQueue srv rcvId rcvPrivateKey sndId decryptKey verifyKey status+    rcvQueue _ = Nothing -retrieveSndQueueByConnAlias_ :: DB.Connection -> ConnAlias -> IO (Maybe SndQueue)-retrieveSndQueueByConnAlias_ dbConn connAlias = do-  r <--    DB.queryNamed+getSndQueueByConnAlias_ :: DB.Connection -> ConnId -> IO (Maybe SndQueue)+getSndQueueByConnAlias_ dbConn connId =+  sndQueue+    <$> DB.query       dbConn       [sql|-        SELECT-          s.key_hash, q.host, q.port, q.snd_id, q.conn_alias,-          q.snd_private_key, q.encrypt_key, q.sign_key, q.status+        SELECT s.key_hash, q.host, q.port, q.snd_id, q.snd_private_key, q.encrypt_key, q.sign_key, q.status         FROM snd_queues q         INNER JOIN servers s ON q.host = s.host AND q.port = s.port-        WHERE q.conn_alias = :conn_alias;+        WHERE q.conn_alias = ?;       |]-      [":conn_alias" := connAlias]-  case r of-    [(keyHash, host, port, sndId, cAlias, sndPrivateKey, encryptKey, signKey, status)] -> do+      (Only connId)+  where+    sndQueue [(keyHash, host, port, sndId, sndPrivateKey, encryptKey, signKey, status)] =       let srv = SMPServer host (deserializePort_ port) keyHash-      return . Just $ SndQueue srv sndId cAlias sndPrivateKey encryptKey signKey status-    _ -> return Nothing+       in Just $ SndQueue srv sndId sndPrivateKey encryptKey signKey status+    sndQueue _ = Nothing  -- * upgradeRcvConnToDuplex helpers -updateConnWithSndQueue_ :: DB.Connection -> ConnAlias -> SndQueue -> IO ()-updateConnWithSndQueue_ dbConn connAlias SndQueue {server, sndId} = do+updateConnWithSndQueue_ :: DB.Connection -> ConnId -> SndQueue -> IO ()+updateConnWithSndQueue_ dbConn connId SndQueue {server, sndId} = do   let port_ = serializePort_ $ port server   DB.executeNamed     dbConn@@ -475,12 +693,12 @@       SET snd_host = :snd_host, snd_port = :snd_port, snd_id = :snd_id       WHERE conn_alias = :conn_alias;     |]-    [":snd_host" := host server, ":snd_port" := port_, ":snd_id" := sndId, ":conn_alias" := connAlias]+    [":snd_host" := host server, ":snd_port" := port_, ":snd_id" := sndId, ":conn_alias" := connId]  -- * upgradeSndConnToDuplex helpers -updateConnWithRcvQueue_ :: DB.Connection -> ConnAlias -> RcvQueue -> IO ()-updateConnWithRcvQueue_ dbConn connAlias RcvQueue {server, rcvId} = do+updateConnWithRcvQueue_ :: DB.Connection -> ConnId -> RcvQueue -> IO ()+updateConnWithRcvQueue_ dbConn connId RcvQueue {server, rcvId} = do   let port_ = serializePort_ $ port server   DB.executeNamed     dbConn@@ -489,12 +707,12 @@       SET rcv_host = :rcv_host, rcv_port = :rcv_port, rcv_id = :rcv_id       WHERE conn_alias = :conn_alias;     |]-    [":rcv_host" := host server, ":rcv_port" := port_, ":rcv_id" := rcvId, ":conn_alias" := connAlias]+    [":rcv_host" := host server, ":rcv_port" := port_, ":rcv_id" := rcvId, ":conn_alias" := connId]  -- * updateRcvIds helpers -retrieveLastIdsAndHashRcv_ :: DB.Connection -> ConnAlias -> IO (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash)-retrieveLastIdsAndHashRcv_ dbConn connAlias = do+retrieveLastIdsAndHashRcv_ :: DB.Connection -> ConnId -> IO (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash)+retrieveLastIdsAndHashRcv_ dbConn connId = do   [(lastInternalId, lastInternalRcvId, lastExternalSndId, lastRcvHash)] <-     DB.queryNamed       dbConn@@ -503,11 +721,11 @@         FROM connections         WHERE conn_alias = :conn_alias;       |]-      [":conn_alias" := connAlias]+      [":conn_alias" := connId]   return (lastInternalId, lastInternalRcvId, lastExternalSndId, lastRcvHash) -updateLastIdsRcv_ :: DB.Connection -> ConnAlias -> InternalId -> InternalRcvId -> IO ()-updateLastIdsRcv_ dbConn connAlias newInternalId newInternalRcvId =+updateLastIdsRcv_ :: DB.Connection -> ConnId -> InternalId -> InternalRcvId -> IO ()+updateLastIdsRcv_ dbConn connId newInternalId newInternalRcvId =   DB.executeNamed     dbConn     [sql|@@ -518,30 +736,32 @@     |]     [ ":last_internal_msg_id" := newInternalId,       ":last_internal_rcv_msg_id" := newInternalRcvId,-      ":conn_alias" := connAlias+      ":conn_alias" := connId     ]  -- * createRcvMsg helpers -insertRcvMsgBase_ :: DB.Connection -> ConnAlias -> RcvMsgData -> IO ()-insertRcvMsgBase_ dbConn connAlias RcvMsgData {..} = do+insertRcvMsgBase_ :: DB.Connection -> ConnId -> RcvMsgData -> IO ()+insertRcvMsgBase_ dbConn connId RcvMsgData {msgMeta, msgBody, internalRcvId} = do+  let MsgMeta {recipient = (internalId, internalTs)} = msgMeta   DB.executeNamed     dbConn     [sql|       INSERT INTO messages-        ( conn_alias, internal_id, internal_ts, internal_rcv_id, internal_snd_id, body)+        ( conn_alias, internal_id, internal_ts, internal_rcv_id, internal_snd_id, body, msg_body)       VALUES-        (:conn_alias,:internal_id,:internal_ts,:internal_rcv_id,            NULL,:body);+        (:conn_alias,:internal_id,:internal_ts,:internal_rcv_id,            NULL,   '',:msg_body);     |]-    [ ":conn_alias" := connAlias,+    [ ":conn_alias" := connId,       ":internal_id" := internalId,       ":internal_ts" := internalTs,       ":internal_rcv_id" := internalRcvId,-      ":body" := decodeUtf8 msgBody+      ":msg_body" := msgBody     ] -insertRcvMsgDetails_ :: DB.Connection -> ConnAlias -> RcvMsgData -> IO ()-insertRcvMsgDetails_ dbConn connAlias RcvMsgData {..} =+insertRcvMsgDetails_ :: DB.Connection -> ConnId -> RcvMsgData -> IO ()+insertRcvMsgDetails_ dbConn connId RcvMsgData {msgMeta, internalRcvId, internalHash, externalPrevSndHash} = do+  let MsgMeta {integrity, recipient, sender, broker} = msgMeta   DB.executeNamed     dbConn     [sql|@@ -554,21 +774,21 @@          :broker_id,:broker_ts,:rcv_status,           NULL,          NULL,          :internal_hash,:external_prev_snd_hash,:integrity);     |]-    [ ":conn_alias" := connAlias,+    [ ":conn_alias" := connId,       ":internal_rcv_id" := internalRcvId,-      ":internal_id" := internalId,-      ":external_snd_id" := fst senderMeta,-      ":external_snd_ts" := snd senderMeta,-      ":broker_id" := fst brokerMeta,-      ":broker_ts" := snd brokerMeta,+      ":internal_id" := fst recipient,+      ":external_snd_id" := fst sender,+      ":external_snd_ts" := snd sender,+      ":broker_id" := fst broker,+      ":broker_ts" := snd broker,       ":rcv_status" := Received,       ":internal_hash" := internalHash,       ":external_prev_snd_hash" := externalPrevSndHash,-      ":integrity" := msgIntegrity+      ":integrity" := integrity     ] -updateHashRcv_ :: DB.Connection -> ConnAlias -> RcvMsgData -> IO ()-updateHashRcv_ dbConn connAlias RcvMsgData {..} =+updateHashRcv_ :: DB.Connection -> ConnId -> RcvMsgData -> IO ()+updateHashRcv_ dbConn connId RcvMsgData {msgMeta, internalHash, internalRcvId} =   DB.executeNamed     dbConn     -- last_internal_rcv_msg_id equality check prevents race condition in case next id was reserved@@ -579,16 +799,16 @@       WHERE conn_alias = :conn_alias         AND last_internal_rcv_msg_id = :last_internal_rcv_msg_id;     |]-    [ ":last_external_snd_msg_id" := fst senderMeta,+    [ ":last_external_snd_msg_id" := fst (sender msgMeta),       ":last_rcv_msg_hash" := internalHash,-      ":conn_alias" := connAlias,+      ":conn_alias" := connId,       ":last_internal_rcv_msg_id" := internalRcvId     ]  -- * updateSndIds helpers -retrieveLastIdsAndHashSnd_ :: DB.Connection -> ConnAlias -> IO (InternalId, InternalSndId, PrevSndMsgHash)-retrieveLastIdsAndHashSnd_ dbConn connAlias = do+retrieveLastIdsAndHashSnd_ :: DB.Connection -> ConnId -> IO (InternalId, InternalSndId, PrevSndMsgHash)+retrieveLastIdsAndHashSnd_ dbConn connId = do   [(lastInternalId, lastInternalSndId, lastSndHash)] <-     DB.queryNamed       dbConn@@ -597,11 +817,11 @@         FROM connections         WHERE conn_alias = :conn_alias;       |]-      [":conn_alias" := connAlias]+      [":conn_alias" := connId]   return (lastInternalId, lastInternalSndId, lastSndHash) -updateLastIdsSnd_ :: DB.Connection -> ConnAlias -> InternalId -> InternalSndId -> IO ()-updateLastIdsSnd_ dbConn connAlias newInternalId newInternalSndId =+updateLastIdsSnd_ :: DB.Connection -> ConnId -> InternalId -> InternalSndId -> IO ()+updateLastIdsSnd_ dbConn connId newInternalId newInternalSndId =   DB.executeNamed     dbConn     [sql|@@ -612,47 +832,48 @@     |]     [ ":last_internal_msg_id" := newInternalId,       ":last_internal_snd_msg_id" := newInternalSndId,-      ":conn_alias" := connAlias+      ":conn_alias" := connId     ]  -- * createSndMsg helpers -insertSndMsgBase_ :: DB.Connection -> ConnAlias -> SndMsgData -> IO ()-insertSndMsgBase_ dbConn connAlias SndMsgData {..} = do+insertSndMsgBase_ :: DB.Connection -> ConnId -> SndMsgData -> IO ()+insertSndMsgBase_ dbConn connId SndMsgData {..} = do   DB.executeNamed     dbConn     [sql|       INSERT INTO messages-        ( conn_alias, internal_id, internal_ts, internal_rcv_id, internal_snd_id, body)+        ( conn_alias, internal_id, internal_ts, internal_rcv_id, internal_snd_id, body, msg_body)       VALUES-        (:conn_alias,:internal_id,:internal_ts,            NULL,:internal_snd_id,:body);+        (:conn_alias,:internal_id,:internal_ts,            NULL,:internal_snd_id,   '',:msg_body);     |]-    [ ":conn_alias" := connAlias,+    [ ":conn_alias" := connId,       ":internal_id" := internalId,       ":internal_ts" := internalTs,       ":internal_snd_id" := internalSndId,-      ":body" := decodeUtf8 msgBody+      ":msg_body" := msgBody     ] -insertSndMsgDetails_ :: DB.Connection -> ConnAlias -> SndMsgData -> IO ()-insertSndMsgDetails_ dbConn connAlias SndMsgData {..} =+insertSndMsgDetails_ :: DB.Connection -> ConnId -> SndMsgData -> IO ()+insertSndMsgDetails_ dbConn connId SndMsgData {..} =   DB.executeNamed     dbConn     [sql|       INSERT INTO snd_messages-        ( conn_alias, internal_snd_id, internal_id, snd_status, sent_ts, delivered_ts, internal_hash)+        ( conn_alias, internal_snd_id, internal_id, snd_status, sent_ts, delivered_ts, internal_hash, previous_msg_hash)       VALUES-        (:conn_alias,:internal_snd_id,:internal_id,:snd_status,    NULL,         NULL,:internal_hash);+        (:conn_alias,:internal_snd_id,:internal_id,:snd_status,    NULL,         NULL,:internal_hash,:previous_msg_hash);     |]-    [ ":conn_alias" := connAlias,+    [ ":conn_alias" := connId,       ":internal_snd_id" := internalSndId,       ":internal_id" := internalId,-      ":snd_status" := Created,-      ":internal_hash" := internalHash+      ":snd_status" := SndMsgCreated,+      ":internal_hash" := internalHash,+      ":previous_msg_hash" := previousMsgHash     ] -updateHashSnd_ :: DB.Connection -> ConnAlias -> SndMsgData -> IO ()-updateHashSnd_ dbConn connAlias SndMsgData {..} =+updateHashSnd_ :: DB.Connection -> ConnId -> SndMsgData -> IO ()+updateHashSnd_ dbConn connId SndMsgData {..} =   DB.executeNamed     dbConn     -- last_internal_snd_msg_id equality check prevents race condition in case next id was reserved@@ -663,6 +884,39 @@         AND last_internal_snd_msg_id = :last_internal_snd_msg_id;     |]     [ ":last_snd_msg_hash" := internalHash,-      ":conn_alias" := connAlias,+      ":conn_alias" := connId,       ":last_internal_snd_msg_id" := internalSndId     ]++-- create record with a random ID++getConnId_ :: DB.Connection -> TVar ChaChaDRG -> ConnData -> IO (Either StoreError ConnId)+getConnId_ dbConn gVar ConnData {connId = ""} = getUniqueRandomId gVar $ getConnData_ dbConn+getConnId_ _ _ ConnData {connId} = pure $ Right connId++getUniqueRandomId :: TVar ChaChaDRG -> (ByteString -> IO (Maybe a)) -> IO (Either StoreError ByteString)+getUniqueRandomId gVar get = tryGet 3+  where+    tryGet :: Int -> IO (Either StoreError ByteString)+    tryGet 0 = pure $ Left SEUniqueID+    tryGet n = do+      id' <- randomId gVar 12+      get id' >>= \case+        Nothing -> pure $ Right id'+        Just _ -> tryGet (n - 1)++createWithRandomId :: TVar ChaChaDRG -> (ByteString -> IO ()) -> IO (Either StoreError ByteString)+createWithRandomId gVar create = tryCreate 3+  where+    tryCreate :: Int -> IO (Either StoreError ByteString)+    tryCreate 0 = pure $ Left SEUniqueID+    tryCreate n = do+      id' <- randomId gVar 12+      E.try (create id') >>= \case+        Right _ -> pure $ Right id'+        Left e+          | DB.sqlError e == DB.ErrorConstraint -> tryCreate (n - 1)+          | otherwise -> pure . Left . SEInternal $ bshow e++randomId :: TVar ChaChaDRG -> Int -> IO ByteString+randomId gVar n = encode <$> (atomically . stateTVar gVar $ randomBytesGenerate n)
+ src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs view
@@ -0,0 +1,72 @@+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TupleSections #-}++module Simplex.Messaging.Agent.Store.SQLite.Migrations+  ( Migration (..),+    app,+    initialize,+    get,+    run,+  )+where++import Control.Monad (forM_)+import Data.FileEmbed (embedDir, makeRelativeToProject)+import Data.Function (on)+import Data.List (intercalate, sortBy)+import Data.Text (Text)+import Data.Text.Encoding (decodeUtf8)+import Data.Time.Clock (getCurrentTime)+import Database.SQLite.Simple (Connection, Only (..))+import qualified Database.SQLite.Simple as DB+import Database.SQLite.Simple.QQ (sql)+import qualified Database.SQLite3 as SQLite3+import System.FilePath (takeBaseName, takeExtension)++data Migration = Migration {name :: String, up :: Text}+  deriving (Show)++-- | The list of migrations in ascending order by date+app :: [Migration]+app =+  sortBy (compare `on` name) . map migration . filter sqlFile $+    $(makeRelativeToProject "migrations" >>= embedDir)+  where+    sqlFile (file, _) = takeExtension file == ".sql"+    migration (file, qStr) = Migration {name = takeBaseName file, up = decodeUtf8 qStr}++get :: Connection -> [Migration] -> IO (Either String [Migration])+get conn migrations =+  migrationsToRun migrations . map fromOnly+    <$> DB.query_ conn "SELECT name FROM migrations ORDER BY name ASC;"++run :: Connection -> [Migration] -> IO ()+run conn ms = DB.withImmediateTransaction conn . forM_ ms $+  \Migration {name, up} -> insert name >> execSQL up+  where+    insert name = DB.execute conn "INSERT INTO migrations (name, ts) VALUES (?, ?);" . (name,) =<< getCurrentTime+    execSQL = SQLite3.exec $ DB.connectionHandle conn++initialize :: Connection -> IO ()+initialize conn =+  DB.execute_+    conn+    [sql|+      CREATE TABLE IF NOT EXISTS migrations (+        name TEXT NOT NULL,+        ts TEXT NOT NULL,+        PRIMARY KEY (name)+      );+    |]++migrationsToRun :: [Migration] -> [String] -> Either String [Migration]+migrationsToRun appMs [] = Right appMs+migrationsToRun [] dbMs = Left $ "database version is newer than the app: " <> intercalate ", " dbMs+migrationsToRun (a : as) (d : ds)+  | name a == d = migrationsToRun as ds+  | otherwise = Left $ "different migration in the app/database: " <> name a <> " / " <> d
− src/Simplex/Messaging/Agent/Store/SQLite/Schema.hs 
@@ -1,167 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE QuasiQuotes #-}--module Simplex.Messaging.Agent.Store.SQLite.Schema (createSchema) where--import Database.SQLite.Simple (Connection, Query, execute_)-import Database.SQLite.Simple.QQ (sql)--createSchema :: Connection -> IO ()-createSchema conn =-  mapM_-    (execute_ conn)-    [ servers,-      rcvQueues,-      sndQueues,-      connections,-      messages,-      rcvMessages,-      sndMessages-    ]---- port is either a port number or a service name, see Network.Socket.Info.ServiceName-servers :: Query-servers =-  [sql|-    CREATE TABLE IF NOT EXISTS servers(-      host TEXT NOT NULL,-      port TEXT NOT NULL,-      key_hash BLOB,-      PRIMARY KEY (host, port)-    ) WITHOUT ROWID;-  |]--rcvQueues :: Query-rcvQueues =-  [sql|-    CREATE TABLE IF NOT EXISTS rcv_queues(-      host TEXT NOT NULL,-      port TEXT NOT NULL,-      rcv_id BLOB NOT NULL,-      conn_alias BLOB NOT NULL,-      rcv_private_key BLOB NOT NULL,-      snd_id BLOB,-      snd_key BLOB,-      decrypt_key BLOB NOT NULL,-      verify_key BLOB,-      status TEXT NOT NULL,-      PRIMARY KEY (host, port, rcv_id),-      FOREIGN KEY (host, port) REFERENCES servers (host, port),-      FOREIGN KEY (conn_alias)-        REFERENCES connections (conn_alias)-        ON DELETE CASCADE-        DEFERRABLE INITIALLY DEFERRED,-      UNIQUE (host, port, snd_id)-    ) WITHOUT ROWID;-  |]--sndQueues :: Query-sndQueues =-  [sql|-    CREATE TABLE IF NOT EXISTS snd_queues(-      host TEXT NOT NULL,-      port TEXT NOT NULL,-      snd_id BLOB NOT NULL,-      conn_alias BLOB NOT NULL,-      snd_private_key BLOB NOT NULL,-      encrypt_key BLOB NOT NULL,-      sign_key BLOB NOT NULL,-      status TEXT NOT NULL,-      PRIMARY KEY (host, port, snd_id),-      FOREIGN KEY (host, port) REFERENCES servers (host, port),-      FOREIGN KEY (conn_alias)-        REFERENCES connections (conn_alias)-        ON DELETE CASCADE-        DEFERRABLE INITIALLY DEFERRED-    ) WITHOUT ROWID;-  |]--connections :: Query-connections =-  [sql|-    CREATE TABLE IF NOT EXISTS connections(-      conn_alias BLOB NOT NULL,-      rcv_host TEXT,-      rcv_port TEXT,-      rcv_id BLOB,-      snd_host TEXT,-      snd_port TEXT,-      snd_id BLOB,-      last_internal_msg_id INTEGER NOT NULL,-      last_internal_rcv_msg_id INTEGER NOT NULL,-      last_internal_snd_msg_id INTEGER NOT NULL,-      last_external_snd_msg_id INTEGER NOT NULL,-      last_rcv_msg_hash BLOB NOT NULL,-      last_snd_msg_hash BLOB NOT NULL,-      PRIMARY KEY (conn_alias),-      FOREIGN KEY (rcv_host, rcv_port, rcv_id) REFERENCES rcv_queues (host, port, rcv_id),-      FOREIGN KEY (snd_host, snd_port, snd_id) REFERENCES snd_queues (host, port, snd_id)-    ) WITHOUT ROWID;-  |]--messages :: Query-messages =-  [sql|-    CREATE TABLE IF NOT EXISTS messages(-      conn_alias BLOB NOT NULL,-      internal_id INTEGER NOT NULL,-      internal_ts TEXT NOT NULL,-      internal_rcv_id INTEGER,-      internal_snd_id INTEGER,-      body TEXT NOT NULL,-      PRIMARY KEY (conn_alias, internal_id),-      FOREIGN KEY (conn_alias)-        REFERENCES connections (conn_alias)-        ON DELETE CASCADE,-      FOREIGN KEY (conn_alias, internal_rcv_id)-        REFERENCES rcv_messages (conn_alias, internal_rcv_id)-        ON DELETE CASCADE-        DEFERRABLE INITIALLY DEFERRED,-      FOREIGN KEY (conn_alias, internal_snd_id)-        REFERENCES snd_messages (conn_alias, internal_snd_id)-        ON DELETE CASCADE-        DEFERRABLE INITIALLY DEFERRED-    ) WITHOUT ROWID;-  |]--rcvMessages :: Query-rcvMessages =-  [sql|-    CREATE TABLE IF NOT EXISTS rcv_messages(-      conn_alias BLOB NOT NULL,-      internal_rcv_id INTEGER NOT NULL,-      internal_id INTEGER NOT NULL,-      external_snd_id INTEGER NOT NULL,-      external_snd_ts TEXT NOT NULL,-      broker_id BLOB NOT NULL,-      broker_ts TEXT NOT NULL,-      rcv_status TEXT NOT NULL,-      ack_brocker_ts TEXT,-      ack_sender_ts TEXT,-      internal_hash BLOB NOT NULL,-      external_prev_snd_hash BLOB NOT NULL,-      integrity BLOB NOT NULL,-      PRIMARY KEY (conn_alias, internal_rcv_id),-      FOREIGN KEY (conn_alias, internal_id)-        REFERENCES messages (conn_alias, internal_id)-        ON DELETE CASCADE-    ) WITHOUT ROWID;-  |]--sndMessages :: Query-sndMessages =-  [sql|-    CREATE TABLE IF NOT EXISTS snd_messages(-      conn_alias BLOB NOT NULL,-      internal_snd_id INTEGER NOT NULL,-      internal_id INTEGER NOT NULL,-      snd_status TEXT NOT NULL,-      sent_ts TEXT,-      delivered_ts TEXT,-      internal_hash BLOB NOT NULL,-      PRIMARY KEY (conn_alias, internal_snd_id),-      FOREIGN KEY (conn_alias, internal_id)-        REFERENCES messages (conn_alias, internal_id)-        ON DELETE CASCADE-    ) WITHOUT ROWID;-  |]
src/Simplex/Messaging/Client.hs view
@@ -53,9 +53,10 @@ import Control.Monad.Trans.Class import Control.Monad.Trans.Except import Data.ByteString.Char8 (ByteString)+import qualified Data.ByteString.Char8 as B import Data.Map.Strict (Map) import qualified Data.Map.Strict as M-import Data.Maybe+import Data.Maybe (fromMaybe) import Network.Socket (ServiceName) import Numeric.Natural import Simplex.Messaging.Agent.Protocol (SMPServer (..))@@ -64,7 +65,7 @@ import Simplex.Messaging.Transport (ATransport (..), TCP, THandle (..), TProxy, Transport (..), TransportError, clientHandshake, runTransportClient) import Simplex.Messaging.Transport.WebSockets (WS) import Simplex.Messaging.Util (bshow, liftError, raceAny_)-import System.Timeout+import System.Timeout (timeout)  -- | 'SMPClient' is a handle used to send commands to a specific SMP server. --@@ -98,6 +99,9 @@     tcpTimeout :: Int,     -- | period for SMP ping commands (microseconds)     smpPing :: Int,+    -- | SMP transport block size, Nothing - the block size will be set by the server.+    -- Allowed sizes are 4, 8, 16, 32, 64 KiB (* 1024 bytes).+    smpBlockSize :: Maybe Int,     -- | estimated maximum size of SMP command excluding message body,     -- determines the maximum allowed message size     smpCommandSize :: Int@@ -111,6 +115,7 @@       defaultTransport = ("5223", transport @TCP),       tcpTimeout = 4_000_000,       smpPing = 30_000_000,+      smpBlockSize = Just 8192,       smpCommandSize = 256     } @@ -127,7 +132,7 @@ -- A single queue can be used for multiple 'SMPClient' instances, -- as 'SMPServerTransmission' includes server information. getSMPClient :: SMPServer -> SMPClientConfig -> TBQueue SMPServerTransmission -> IO () -> IO (Either SMPClientError SMPClient)-getSMPClient smpServer cfg@SMPClientConfig {qSize, tcpTimeout, smpPing} msgQ disconnected =+getSMPClient smpServer cfg@SMPClientConfig {qSize, tcpTimeout, smpPing, smpBlockSize} msgQ disconnected =   atomically mkSMPClient >>= runClient useTransport   where     mkSMPClient :: STM SMPClient@@ -172,7 +177,7 @@      client :: forall c. Transport c => TProxy c -> SMPClient -> TMVar (Either SMPClientError Int) -> c -> IO ()     client _ c thVar h =-      runExceptT (clientHandshake h $ keyHash smpServer) >>= \case+      runExceptT (clientHandshake h smpBlockSize $ keyHash smpServer) >>= \case         Left e -> atomically . putTMVar thVar . Left $ SMPTransportError e         Right th -> do           atomically $ do@@ -195,22 +200,27 @@     process :: SMPClient -> IO ()     process SMPClient {rcvQ, sentCommands} = forever $ do       (_, (corrId, qId, respOrErr)) <- atomically $ readTBQueue rcvQ-      cs <- readTVarIO sentCommands-      case M.lookup corrId cs of-        Nothing -> do-          case respOrErr of-            Right (Cmd SBroker cmd) -> atomically $ writeTBQueue msgQ (smpServer, qId, cmd)-            -- TODO send everything else to errQ and log in agent-            _ -> return ()-        Just Request {queueId, responseVar} -> atomically $ do-          modifyTVar sentCommands $ M.delete corrId-          putTMVar responseVar $-            if queueId == qId-              then case respOrErr of-                Left e -> Left $ SMPResponseError e-                Right (Cmd _ (ERR e)) -> Left $ SMPServerError e-                Right r -> Right r-              else Left SMPUnexpectedResponse+      if B.null $ bs corrId+        then sendMsg qId respOrErr+        else do+          cs <- readTVarIO sentCommands+          case M.lookup corrId cs of+            Nothing -> sendMsg qId respOrErr+            Just Request {queueId, responseVar} -> atomically $ do+              modifyTVar sentCommands $ M.delete corrId+              putTMVar responseVar $+                if queueId == qId+                  then case respOrErr of+                    Left e -> Left $ SMPResponseError e+                    Right (Cmd _ (ERR e)) -> Left $ SMPServerError e+                    Right r -> Right r+                  else Left SMPUnexpectedResponse++    sendMsg :: QueueId -> Either ErrorType Cmd -> IO ()+    sendMsg qId = \case+      Right (Cmd SBroker cmd) -> atomically $ writeTBQueue msgQ (smpServer, qId, cmd)+      -- TODO send everything else to errQ and log in agent+      _ -> return ()  -- | Disconnects SMP client from the server and terminates client threads. closeSMPClient :: SMPClient -> IO ()
src/Simplex/Messaging/Crypto.hs view
@@ -20,18 +20,20 @@ -- <https://hackage.haskell.org/package/cryptonite cryptonite package>. module Simplex.Messaging.Crypto   ( -- * RSA keys-    PrivateKey (rsaPrivateKey),-    SafePrivateKey, -- constructor is not exported+    PrivateKey (rsaPrivateKey, publicKey),+    SafePrivateKey (..), -- constructor is not exported     FullPrivateKey (..),+    APrivateKey (..),     PublicKey (..),     SafeKeyPair,     FullKeyPair,     KeyHash (..),     generateKeyPair,-    publicKey,+    publicKey',     publicKeySize,     validKeySize,     safePrivateKey,+    removePublicKey,      -- * E2E hybrid encryption scheme     encrypt,@@ -121,6 +123,9 @@ -- | A newtype of 'Crypto.PubKey.RSA.PrivateKey' (with PublicKey inside). newtype FullPrivateKey = FullPrivateKey {unPrivateKey :: R.PrivateKey} deriving (Eq, Show) +-- | A newtype of 'Crypto.PubKey.RSA.PrivateKey' (PublicKey may be inside).+newtype APrivateKey = APrivateKey {unPrivateKey :: R.PrivateKey} deriving (Eq, Show)+ -- | Type-class used for both private key types: SafePrivateKey and FullPrivateKey. class PrivateKey k where   -- unwraps 'Crypto.PubKey.RSA.PrivateKey'@@ -132,29 +137,53 @@   -- smart constructor removing public key from SafePrivateKey but keeping it in FullPrivateKey   mkPrivateKey :: R.PrivateKey -> k +  -- extracts public key from private key+  publicKey :: k -> Maybe PublicKey++-- | Remove public key exponent from APrivateKey.+removePublicKey :: APrivateKey -> APrivateKey+removePublicKey (APrivateKey R.PrivateKey {private_pub = k, private_d}) =+  APrivateKey $ unPrivateKey (safePrivateKey (R.public_size k, R.public_n k, private_d) :: SafePrivateKey)+ instance PrivateKey SafePrivateKey where   rsaPrivateKey = unPrivateKey   _privateKey = SafePrivateKey   mkPrivateKey R.PrivateKey {private_pub = k, private_d} =     safePrivateKey (R.public_size k, R.public_n k, private_d)+  publicKey _ = Nothing  instance PrivateKey FullPrivateKey where   rsaPrivateKey = unPrivateKey   _privateKey = FullPrivateKey   mkPrivateKey = FullPrivateKey+  publicKey = Just . PublicKey . R.private_pub . rsaPrivateKey +instance PrivateKey APrivateKey where+  rsaPrivateKey = unPrivateKey+  _privateKey = APrivateKey+  mkPrivateKey = APrivateKey+  publicKey pk =+    let k = R.private_pub $ rsaPrivateKey pk+     in if R.public_e k == 0+          then Nothing+          else Just $ PublicKey k+ instance IsString FullPrivateKey where-  fromString = parseString (decode >=> decodePrivKey)+  fromString = parseString $ decode >=> decodePrivKey  instance IsString PublicKey where-  fromString = parseString (decode >=> decodePubKey)+  fromString = parseString $ decode >=> decodePubKey  instance ToField SafePrivateKey where toField = toField . encodePrivKey +instance ToField APrivateKey where toField = toField . encodePrivKey+ instance ToField PublicKey where toField = toField . encodePubKey  instance FromField SafePrivateKey where fromField = blobFieldParser binaryPrivKeyP +instance FromField APrivateKey where fromField = blobFieldParser binaryPrivKeyP+ instance FromField PublicKey where fromField = blobFieldParser binaryPubKeyP  -- | Tuple of RSA 'PublicKey' and 'PrivateKey'.@@ -217,8 +246,8 @@ privateKeySize :: PrivateKey k => k -> Int privateKeySize = R.public_size . R.private_pub . rsaPrivateKey -publicKey :: FullPrivateKey -> PublicKey-publicKey = PublicKey . R.private_pub . rsaPrivateKey+publicKey' :: FullPrivateKey -> PublicKey+publicKey' = PublicKey . R.private_pub . rsaPrivateKey  publicKeySize :: PublicKey -> Int publicKeySize = R.public_size . rsaPublicKey@@ -227,6 +256,7 @@ validKeySize = \case   128 -> True   256 -> True+  384 -> True   512 -> True   _ -> False
src/Simplex/Messaging/Parsers.hs view
@@ -30,7 +30,7 @@   pure $ str <> pad  tsISO8601P :: Parser UTCTime-tsISO8601P = maybe (fail "timestamp") pure . parseISO8601 . B.unpack =<< A.takeTill (== ' ')+tsISO8601P = maybe (fail "timestamp") pure . parseISO8601 . B.unpack =<< A.takeTill wordEnd  parse :: Parser a -> e -> (ByteString -> Either e a) parse parser err = first (const err) . parseAll parser@@ -42,13 +42,16 @@ parseRead = (>>= maybe (fail "cannot read") pure . readMaybe . B.unpack)  parseRead1 :: Read a => Parser a-parseRead1 = parseRead $ A.takeTill (== ' ')+parseRead1 = parseRead $ A.takeTill wordEnd  parseRead2 :: Read a => Parser a parseRead2 = parseRead $ do-  w1 <- A.takeTill (== ' ') <* A.char ' '-  w2 <- A.takeTill (== ' ')+  w1 <- A.takeTill wordEnd <* A.char ' '+  w2 <- A.takeTill wordEnd   pure $ w1 <> " " <> w2++wordEnd :: Char -> Bool+wordEnd c = c == ' ' || c == '\n'  parseString :: (ByteString -> Either String a) -> (String -> a) parseString p = either error id . p . B.pack
src/Simplex/Messaging/Protocol.hs view
@@ -192,6 +192,8 @@     CMD CommandError   | -- | command authorization error - bad signature or non-existing SMP queue     AUTH+  | -- | SMP queue capacity is exceeded on the server+    QUOTA   | -- | ACK command is sent without message to be acknowledged     NO_MSG   | -- | internal server error
src/Simplex/Messaging/Server.hs view
@@ -90,7 +90,8 @@ runClient :: (Transport c, MonadUnliftIO m, MonadReader Env m) => TProxy c -> c -> m () runClient _ h = do   keyPair <- asks serverKeyPair-  liftIO (runExceptT $ serverHandshake h keyPair) >>= \case+  ServerConfig {blockSize} <- asks config+  liftIO (runExceptT $ serverHandshake h blockSize keyPair) >>= \case     Right th -> runClientTransport th     Left _ -> pure () @@ -157,6 +158,7 @@       cryptoVerify $ case sigLen of         128 -> dummyKey128         256 -> dummyKey256+        384 -> dummyKey384         512 -> dummyKey512         _ -> dummyKey256     sigLen = B.length $ C.unSignature sig@@ -169,6 +171,9 @@ dummyKey256 :: C.PublicKey dummyKey256 = "MIIBoDANBgkqhkiG9w0BAQEFAAOCAY0AMIIBiAKCAQEAxwmTvaqmdTbkfUGNi8Yu0L/T4cxuOlQlx3zGZ9X9Qx0+oZjknWK+QHrdWTcpS+zH4Hi7fP6kanOQoQ90Hj6Ghl57VU1GEdUPywSw4i1/7t0Wv9uT9Q2ktHp2rqVo3xkC9IVIpL7EZAxdRviIN2OsOB3g4a/F1ZpjxcAaZeOMUugiAX1+GtkLuE0Xn4neYjCaOghLxQTdhybN70VtnkiQLx/X9NjkDIl/spYGm3tQFMyYKkP6IWoEpj0926hJ0fmlmhy8tAOhlZsb/baW5cgkEZ3E9jVVrySCgQzoLQgma610FIISRpRJbSyv26jU7MkMxiyuBiDaFOORkXFttoKbtQKBgEbDS9II2brsz+vfI7uP8atFcawkE52cx4M1UWQhqb1H3tBiRl+qO+dMq1pPQF2bW7dlZAWYzS4W/367bTAuALHBDGB8xi1P4Njhh9vaOgTvuqrHG9NJQ85BLy0qGw8rjIWSIXVmVpfrXFJ8po5l04UE258Ll2yocv3QRQmddQW9" +dummyKey384 :: C.PublicKey+dummyKey384 = "MIICITANBgkqhkiG9w0BAQEFAAOCAg4AMIICCQKCAYEAthExp77lSFBMB0RedjgKIU+oNH5lMGdMqDCG0E5Ly7X49rFpfDMMN08GDIgvzg9kcwV3ScbPcjUE19wmAShX9f9k3w38KM3wmIBKSiuCREQl0V3xAYp1SYwiAkMNSSwxuIkDEeSOR56WdEcZvqbB4lY9MQlUv70KriPDxZaqKCTKslUezXHQuYPQX6eMnGFK7hxz5Kl5MajV52d+5iXsa8CA+m/e1KVnbelCO+xhN89xG8ALt0CJ9k5Wwo3myLgXi4dmNankCmg8jkh+7y2ywkzxMwH1JydDtV/FLzkbZsbPR2w93TNrTq1RJOuqMyh0VtdBSpxNW/Ft988TkkX2BAWzx82INw7W6/QbHGNtHNB995R4sgeYy8QbEpNGBhQnfQh7yRWygLTVXWKApQzzfCeIoDDWUS7dMv/zXoasAnpDBj+6UhHv3BHrps7kBvRyZQ2d/nUuAqiGd43ljJ++n6vNyFLgZoiV7HLia/FOGMkdt7j92CNmFHxiT6Xl7kRHAoGBAPNoWny2O7LBxzAKMLmQVHBAiKp6RMx+7URvtQDHDHPaZ7F3MvtvmYWwGzund3cQFAaV1EkJoYeI3YRuj6xdXgMyMaP54On++btArb6jUtZuvlC98qE8dEEHQNh+7TsCiMU+ivbeKFxS9A/B7OVedoMnPoJWhatbA9zB/6L1GNPh"+ dummyKey512 :: C.PublicKey dummyKey512 = "MIICoDANBgkqhkiG9w0BAQEFAAOCAo0AMIICiAKCAgEArkCY9DuverJ4mmzDektv9aZMFyeRV46WZK9NsOBKEc+1ncqMs+LhLti9asKNgUBRbNzmbOe0NYYftrUpwnATaenggkTFxxbJ4JGJuGYbsEdFWkXSvrbWGtM8YUmn5RkAGme12xQ89bSM4VoJAGnrYPHwmcQd+KYCPZvTUsxaxgrJTX65ejHN9BsAn8XtGViOtHTDJO9yUMD2WrJvd7wnNa+0ugEteDLzMU++xS98VC+uA1vfauUqi3yXVchdfrLdVUuM+JE0gUEXCgzjuHkaoHiaGNiGhdPYoAJJdOKQOIHAKdk7Th6OPhirPhc9XYNB4O8JDthKhNtfokvFIFlC4QBRzJhpLIENaEBDt08WmgpOnecZB/CuxkqqOrNa8j5K5jNrtXAI67W46VEC2jeQy/gZwb64Zit2A4D00xXzGbQTPGj4ehcEMhLx5LSCygViEf0w0tN3c3TEyUcgPzvECd2ZVpQLr9Z4a07Ebr+YSuxcHhjg4Rg1VyJyOTTvaCBGm5X2B3+tI4NUttmikIHOYpBnsLmHY2BgfH2KcrIsDyAhInXmTFr/L2+erFarUnlfATd2L8Ti43TNHDedO6k6jI5Gyi62yPwjqPLEIIK8l+pIeNfHJ3pPmjhHBfzFcQLMMMXffHWNK8kWklrQXK+4j4HiPcTBvlO1FEtG9nEIZhUCgYA4a6WtI2k5YNli1C89GY5rGUY7RP71T6RWri/D3Lz9T7GvU+FemAyYmsvCQwqijUOur0uLvwSP8VdxpSUcrjJJSWur2hrPWzWlu0XbNaeizxpFeKbQP+zSrWJ1z8RwfAeUjShxt8q1TuqGqY10wQyp3nyiTGvS+KwZVj5h5qx8NQ==" @@ -292,16 +297,19 @@               QueueActive -> do                 ms <- asks msgStore                 msg <- mkMessage+                quota <- asks $ msgQueueQuota . config                 atomically $ do-                  q <- getMsgQueue ms (recipientId qr)-                  writeMsg q msg-                  return ok+                  q <- getMsgQueue ms (recipientId qr) quota+                  isFull q >>= \case+                    False -> writeMsg q msg $> ok+                    True -> pure $ err QUOTA          deliverMessage :: (MsgQueue -> STM (Maybe Message)) -> RecipientId -> Sub -> m Transmission         deliverMessage tryPeek rId = \case           Sub {subThread = NoSub} -> do             ms <- asks msgStore-            q <- atomically $ getMsgQueue ms rId+            quota <- asks $ msgQueueQuota . config+            q <- atomically $ getMsgQueue ms rId quota             atomically (tryPeek q) >>= \case               Nothing -> forkSub q $> ok               Just msg -> atomically setDelivered $> mkResp corrId rId (msgCmd msg)
src/Simplex/Messaging/Server/Env/STM.hs view
@@ -25,9 +25,11 @@ data ServerConfig = ServerConfig   { transports :: [(ServiceName, ATransport)],     tbqSize :: Natural,+    msgQueueQuota :: Natural,     queueIdBytes :: Int,     msgIdBytes :: Int,     storeLog :: Maybe (StoreLog 'ReadMode),+    blockSize :: Int,     serverPrivateKey :: C.FullPrivateKey     -- serverId :: ByteString   }@@ -86,7 +88,7 @@   idsDrg <- drgNew >>= newTVarIO   s' <- restoreQueues queueStore `mapM` storeLog (config :: ServerConfig)   let pk = serverPrivateKey config-      serverKeyPair = (C.publicKey pk, pk)+      serverKeyPair = (C.publicKey' pk, pk)   return Env {config, server, queueStore, msgStore, idsDrg, serverKeyPair, storeLog = s'}   where     restoreQueues :: QueueStore -> StoreLog 'ReadMode -> m (StoreLog 'WriteMode)
src/Simplex/Messaging/Server/MsgStore.hs view
@@ -3,6 +3,7 @@ module Simplex.Messaging.Server.MsgStore where  import Data.Time.Clock+import Numeric.Natural import Simplex.Messaging.Protocol (Encoded, MsgBody, RecipientId)  data Message = Message@@ -12,10 +13,11 @@   }  class MonadMsgStore s q m | s -> q where-  getMsgQueue :: s -> RecipientId -> m q+  getMsgQueue :: s -> RecipientId -> Natural -> m q   delMsgQueue :: s -> RecipientId -> m ()  class MonadMsgQueue q m where+  isFull :: q -> m Bool   writeMsg :: q -> Message -> m () -- non blocking   tryPeekMsg :: q -> m (Maybe Message) -- non blocking   peekMsg :: q -> m Message -- blocking
src/Simplex/Messaging/Server/MsgStore/STM.hs view
@@ -8,11 +8,12 @@  import Data.Map.Strict (Map) import qualified Data.Map.Strict as M+import Numeric.Natural import Simplex.Messaging.Protocol (RecipientId) import Simplex.Messaging.Server.MsgStore import UnliftIO.STM -newtype MsgQueue = MsgQueue {msgQueue :: TQueue Message}+newtype MsgQueue = MsgQueue {msgQueue :: TBQueue Message}  newtype MsgStoreData = MsgStoreData {messages :: Map RecipientId MsgQueue} @@ -22,13 +23,13 @@ newMsgStore = newTVar $ MsgStoreData M.empty  instance MonadMsgStore STMMsgStore MsgQueue STM where-  getMsgQueue :: STMMsgStore -> RecipientId -> STM MsgQueue-  getMsgQueue store rId = do+  getMsgQueue :: STMMsgStore -> RecipientId -> Natural -> STM MsgQueue+  getMsgQueue store rId quota = do     m <- messages <$> readTVar store     maybe (newQ m) return $ M.lookup rId m     where       newQ m' = do-        q <- MsgQueue <$> newTQueue+        q <- MsgQueue <$> newTBQueue quota         writeTVar store . MsgStoreData $ M.insert rId q m'         return q @@ -37,15 +38,18 @@     modifyTVar store $ MsgStoreData . M.delete rId . messages  instance MonadMsgQueue MsgQueue STM where+  isFull :: MsgQueue -> STM Bool+  isFull = isFullTBQueue . msgQueue+   writeMsg :: MsgQueue -> Message -> STM ()-  writeMsg = writeTQueue . msgQueue+  writeMsg = writeTBQueue . msgQueue    tryPeekMsg :: MsgQueue -> STM (Maybe Message)-  tryPeekMsg = tryPeekTQueue . msgQueue+  tryPeekMsg = tryPeekTBQueue . msgQueue    peekMsg :: MsgQueue -> STM Message-  peekMsg = peekTQueue . msgQueue+  peekMsg = peekTBQueue . msgQueue    -- atomic delete (== read) last and peek next message if available   tryDelPeekMsg :: MsgQueue -> STM (Maybe Message)-  tryDelPeekMsg (MsgQueue q) = tryReadTQueue q >> tryPeekTQueue q+  tryDelPeekMsg (MsgQueue q) = tryReadTBQueue q >> tryPeekTBQueue q
src/Simplex/Messaging/Transport.hs view
@@ -63,8 +63,10 @@ import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B import Data.Functor (($>))+import Data.Maybe(fromMaybe) import Data.Set (Set) import qualified Data.Set as S+import Data.String import Data.Word (Word32) import GHC.Generics (Generic) import GHC.IO.Exception (IOErrorType (..))@@ -73,7 +75,7 @@ import Network.Socket import Network.Transport.Internal (decodeNum16, decodeNum32, encodeEnum16, encodeEnum32, encodeWord32) import qualified Simplex.Messaging.Crypto as C-import Simplex.Messaging.Parsers (parse, parseAll, parseRead1)+import Simplex.Messaging.Parsers (parse, parseAll, parseRead1, parseString) import Simplex.Messaging.Util (bshow, liftError) import System.IO import System.IO.Error@@ -212,11 +214,14 @@ data SMPVersion = SMPVersion Int Int Int Int   deriving (Eq, Ord) +instance IsString SMPVersion where+  fromString = parseString $ parseAll smpVersionP+ major :: SMPVersion -> (Int, Int) major (SMPVersion a b _ _) = (a, b)  currentSMPVersion :: SMPVersion-currentSMPVersion = SMPVersion 0 3 2 0+currentSMPVersion = "0.4.0.0"  serializeSMPVersion :: SMPVersion -> ByteString serializeSMPVersion (SMPVersion a b c d) = B.intercalate "." [bshow a, bshow b, bshow c, bshow d]@@ -340,21 +345,21 @@ -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#appendix-a -- -- The numbers in function names refer to the steps in the document.-serverHandshake :: forall c. Transport c => c -> C.FullKeyPair -> ExceptT TransportError IO (THandle c)-serverHandshake c (k, pk) = do+serverHandshake :: forall c. Transport c => c -> Int -> C.FullKeyPair -> ExceptT TransportError IO (THandle c)+serverHandshake c srvBlockSize (k, pk) = do+  checkValidBlockSize srvBlockSize   liftIO sendHeaderAndPublicKey_1   encryptedKeys <- receiveEncryptedKeys_4-  -- TODO server currently ignores blockSize returned by the client-  -- this is reserved for future support of streams-  ClientHandshake {blockSize = _, sndKey, rcvKey} <- decryptParseKeys_5 encryptedKeys-  th <- liftIO $ transportHandle c rcvKey sndKey transportBlockSize -- keys are swapped here+  ClientHandshake {blockSize, sndKey, rcvKey} <- decryptParseKeys_5 encryptedKeys+  checkValidBlockSize blockSize+  th <- liftIO $ transportHandle c rcvKey sndKey blockSize -- keys are swapped here   sendWelcome_6 th   pure th   where     sendHeaderAndPublicKey_1 :: IO ()     sendHeaderAndPublicKey_1 = do       let sKey = C.encodePubKey k-          header = ServerHeader {blockSize = transportBlockSize, keySize = B.length sKey}+          header = ServerHeader {blockSize = srvBlockSize, keySize = B.length sKey}       cPut c $ binaryServerHeader header       cPut c sKey     receiveEncryptedKeys_4 :: ExceptT TransportError IO ByteString@@ -374,13 +379,14 @@ -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#appendix-a -- -- The numbers in function names refer to the steps in the document.-clientHandshake :: forall c. Transport c => c -> Maybe C.KeyHash -> ExceptT TransportError IO (THandle c)-clientHandshake c keyHash = do+clientHandshake :: forall c. Transport c => c -> Maybe Int -> Maybe C.KeyHash -> ExceptT TransportError IO (THandle c)+clientHandshake c blkSize_ keyHash = do+  mapM_ checkValidBlockSize blkSize_   (k, blkSize) <- getHeaderAndPublicKey_1_2-  -- TODO currently client always uses the blkSize returned by the server-  keys@ClientHandshake {sndKey, rcvKey} <- liftIO $ generateKeys_3 blkSize-  sendEncryptedKeys_4 k keys-  th <- liftIO $ transportHandle c sndKey rcvKey blkSize+  let clientBlkSize = fromMaybe blkSize blkSize_+  chs@ClientHandshake {sndKey, rcvKey} <- liftIO $ generateKeys_3 clientBlkSize+  sendEncryptedKeys_4 k chs+  th <- liftIO $ transportHandle c sndKey rcvKey clientBlkSize   getWelcome_6 th >>= checkVersion   pure th   where@@ -388,8 +394,7 @@     getHeaderAndPublicKey_1_2 = do       header <- liftIO (cGet c serverHeaderSize)       ServerHeader {blockSize, keySize} <- liftEither $ parse serverHeaderP (TEHandshake HEADER) header-      when (blockSize < transportBlockSize || blockSize > maxTransportBlockSize) $-        throwError $ TEHandshake HEADER+      checkValidBlockSize blockSize       s <- liftIO $ cGet c keySize       maybe (pure ()) (validateKeyHash_2 s) keyHash       key <- liftEither $ parseKey s@@ -408,8 +413,8 @@       baseIV <- C.randomIV       pure SessionKey {aesKey, baseIV, counter = undefined}     sendEncryptedKeys_4 :: C.PublicKey -> ClientHandshake -> ExceptT TransportError IO ()-    sendEncryptedKeys_4 k keys =-      liftError (const $ TEHandshake ENCRYPT) (C.encryptOAEP k $ serializeClientHandshake keys)+    sendEncryptedKeys_4 k chs =+      liftError (const $ TEHandshake ENCRYPT) (C.encryptOAEP k $ serializeClientHandshake chs)         >>= liftIO . cPut c     getWelcome_6 :: THandle c -> ExceptT TransportError IO SMPVersion     getWelcome_6 th = ExceptT $ (>>= parseSMPVersion) <$> tGetEncrypted th@@ -420,17 +425,18 @@       when (major smpVersion > major currentSMPVersion) . throwE $         TEHandshake MAJOR_VERSION +checkValidBlockSize :: Int -> ExceptT TransportError IO ()+checkValidBlockSize blkSize =+  when (blkSize `notElem` transportBlockSizes) . throwError $ TEHandshake HEADER+ data ServerHeader = ServerHeader {blockSize :: Int, keySize :: Int}   deriving (Eq, Show)  binaryRsaTransport :: Int binaryRsaTransport = 0 -transportBlockSize :: Int-transportBlockSize = 4096--maxTransportBlockSize :: Int-maxTransportBlockSize = 65536+transportBlockSizes :: [Int]+transportBlockSizes = map (* 1024) [4, 8, 16, 32, 64]  serverHeaderSize :: Int serverHeaderSize = 8
src/Simplex/Messaging/Util.hs view
@@ -50,3 +50,6 @@  liftEitherError :: (MonadIO m, MonadError e' m) => (e -> e') -> IO (Either e a) -> m a liftEitherError f a = liftIOEither (first f <$> a)++tryError :: MonadError e m => m a -> m (Either e a)+tryError action = (Right <$> action) `catchError` (pure . Left)
tests/AgentTests.hs view
@@ -2,44 +2,64 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE GADTs #-} {-# LANGUAGE LambdaCase #-}-{-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE PatternSynonyms #-}+{-# LANGUAGE PostfixOperators #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS_GHC -fno-warn-incomplete-uni-patterns #-} -module AgentTests where+module AgentTests (agentTests) where +import AgentTests.FunctionalAPITests (functionalAPITests) import AgentTests.SQLiteTests (storeTests) import Control.Concurrent import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B import SMPAgentClient+import SMPClient (testPort, testPort2, testStoreLogFile, withSmpServer, withSmpServerStoreLogOn) import Simplex.Messaging.Agent.Protocol import Simplex.Messaging.Protocol (ErrorType (..), MsgBody) import Simplex.Messaging.Transport (ATransport (..), TProxy (..), Transport (..))+import System.Directory (removeFile) import System.Timeout import Test.Hspec  agentTests :: ATransport -> Spec agentTests (ATransport t) = do+  describe "Functional API" $ functionalAPITests (ATransport t)   describe "SQLite store" storeTests   describe "SMP agent protocol syntax" $ syntaxTests t   describe "Establishing duplex connection" do     it "should connect via one server and one agent" $       smpAgentTest2_1_1 $ testDuplexConnection t+    it "should connect via one server and one agent (random IDs)" $+      smpAgentTest2_1_1 $ testDuplexConnRandomIds t     it "should connect via one server and 2 agents" $       smpAgentTest2_2_1 $ testDuplexConnection t+    it "should connect via one server and 2 agents (random IDs)" $+      smpAgentTest2_2_1 $ testDuplexConnRandomIds t     it "should connect via 2 servers and 2 agents" $       smpAgentTest2_2_2 $ testDuplexConnection t+    it "should connect via 2 servers and 2 agents (random IDs)" $+      smpAgentTest2_2_2 $ testDuplexConnRandomIds t   describe "Connection subscriptions" do     it "should connect via one server and one agent" $       smpAgentTest3_1_1 $ testSubscription t     it "should send notifications to client when server disconnects" $       smpAgentServerTest $ testSubscrNotification t+  describe "Message delivery" do+    it "should deliver messages after losing server connection and re-connecting" $+      smpAgentTest2_2_2_needs_server $ testMsgDeliveryServerRestart t+    it "should deliver pending messages after agent restarting" $+      smpAgentTest1_1_1 $ testMsgDeliveryAgentRestart t +-- | receive message to handle `h`+(<#:) :: Transport c => c -> IO (ATransmissionOrError 'Agent)+(<#:) = tGet SAgent+ -- | send transmission `t` to handle `h` and get response (#:) :: Transport c => c -> (ByteString, ByteString, ByteString) -> IO (ATransmissionOrError 'Agent)-h #: t = tPutRaw h t >> tGet SAgent h+h #: t = tPutRaw h t >> (<#:) h  -- | action and expected response -- `h #:t #> r` is the test that sends `t` to `h` and validates that the response is `r`@@ -58,11 +78,11 @@  -- | receive message to handle `h` and validate that it is the expected one (<#) :: Transport c => c -> ATransmission 'Agent -> Expectation-h <# (corrId, cAlias, cmd) = tGet SAgent h `shouldReturn` (corrId, cAlias, Right cmd)+h <# (corrId, cAlias, cmd) = (h <#:) `shouldReturn` (corrId, cAlias, Right cmd)  -- | receive message to handle `h` and validate it using predicate `p` (<#=) :: Transport c => c -> (ATransmission 'Agent -> Bool) -> Expectation-h <#= p = tGet SAgent h >>= (`shouldSatisfy` p . correctTransmission)+h <#= p = (h <#:) >>= (`shouldSatisfy` p . correctTransmission)  -- | test that nothing is delivered to handle `h` during 10ms (#:#) :: Transport c => c -> String -> Expectation@@ -74,50 +94,183 @@         _ -> return ()  pattern Msg :: MsgBody -> ACommand 'Agent-pattern Msg msgBody <- MSG {msgBody, msgIntegrity = MsgOk}+pattern Msg msgBody <- MSG MsgMeta {integrity = MsgOk} msgBody  testDuplexConnection :: Transport c => TProxy c -> c -> c -> IO () testDuplexConnection _ alice bob = do   ("1", "bob", Right (INV qInfo)) <- alice #: ("1", "bob", "NEW")   let qInfo' = serializeSmpQueueInfo qInfo-  bob #: ("11", "alice", "JOIN " <> qInfo') #> ("", "alice", CON)+  bob #: ("11", "alice", "JOIN " <> qInfo' <> " 14\nbob's connInfo") #> ("11", "alice", OK)+  ("", "bob", Right (REQ confId "bob's connInfo")) <- (alice <#:)+  alice #: ("2", "bob", "ACPT " <> confId <> " 16\nalice's connInfo") #> ("2", "bob", OK)+  bob <# ("", "alice", INFO "alice's connInfo")+  bob <# ("", "alice", CON)   alice <# ("", "bob", CON)-  alice #: ("2", "bob", "SEND :hello") =#> \case ("2", "bob", SENT 1) -> True; _ -> False-  alice #: ("3", "bob", "SEND :how are you?") =#> \case ("3", "bob", SENT 2) -> True; _ -> False+  alice #: ("3", "bob", "SEND :hello") #> ("3", "bob", MID 1)+  alice <# ("", "bob", SENT 1)   bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False+  bob #: ("12", "alice", "ACK 1") #> ("12", "alice", OK)+  alice #: ("4", "bob", "SEND :how are you?") #> ("4", "bob", MID 2)+  alice <# ("", "bob", SENT 2)   bob <#= \case ("", "alice", Msg "how are you?") -> True; _ -> False-  bob #: ("14", "alice", "SEND 9\nhello too") =#> \case ("14", "alice", SENT 3) -> True; _ -> False+  bob #: ("13", "alice", "ACK 2") #> ("13", "alice", OK)+  bob #: ("14", "alice", "SEND 9\nhello too") #> ("14", "alice", MID 3)+  bob <# ("", "alice", SENT 3)   alice <#= \case ("", "bob", Msg "hello too") -> True; _ -> False-  bob #: ("15", "alice", "SEND 9\nmessage 1") =#> \case ("15", "alice", SENT 4) -> True; _ -> False+  alice #: ("3a", "bob", "ACK 3") #> ("3a", "bob", OK)+  bob #: ("15", "alice", "SEND 9\nmessage 1") #> ("15", "alice", MID 4)+  bob <# ("", "alice", SENT 4)   alice <#= \case ("", "bob", Msg "message 1") -> True; _ -> False+  alice #: ("4a", "bob", "ACK 4") #> ("4a", "bob", OK)   alice #: ("5", "bob", "OFF") #> ("5", "bob", OK)-  bob #: ("17", "alice", "SEND 9\nmessage 3") #> ("17", "alice", ERR (SMP AUTH))+  bob #: ("17", "alice", "SEND 9\nmessage 3") #> ("17", "alice", MID 5)+  bob <# ("", "alice", MERR 5 (SMP AUTH))   alice #: ("6", "bob", "DEL") #> ("6", "bob", OK)   alice #:# "nothing else should be delivered to alice" +testDuplexConnRandomIds :: Transport c => TProxy c -> c -> c -> IO ()+testDuplexConnRandomIds _ alice bob = do+  ("1", bobConn, Right (INV qInfo)) <- alice #: ("1", "", "NEW")+  let qInfo' = serializeSmpQueueInfo qInfo+  ("11", aliceConn, Right OK) <- bob #: ("11", "", "JOIN " <> qInfo' <> " 14\nbob's connInfo")+  ("", bobConn', Right (REQ confId "bob's connInfo")) <- (alice <#:)+  bobConn' `shouldBe` bobConn+  alice #: ("2", bobConn, "ACPT " <> confId <> " 16\nalice's connInfo") =#> \case ("2", c, OK) -> c == bobConn; _ -> False+  bob <# ("", aliceConn, INFO "alice's connInfo")+  bob <# ("", aliceConn, CON)+  alice <# ("", bobConn, CON)+  alice #: ("2", bobConn, "SEND :hello") #> ("2", bobConn, MID 1)+  alice <# ("", bobConn, SENT 1)+  bob <#= \case ("", c, Msg "hello") -> c == aliceConn; _ -> False+  bob #: ("12", aliceConn, "ACK 1") #> ("12", aliceConn, OK)+  alice #: ("3", bobConn, "SEND :how are you?") #> ("3", bobConn, MID 2)+  alice <# ("", bobConn, SENT 2)+  bob <#= \case ("", c, Msg "how are you?") -> c == aliceConn; _ -> False+  bob #: ("13", aliceConn, "ACK 2") #> ("13", aliceConn, OK)+  bob #: ("14", aliceConn, "SEND 9\nhello too") #> ("14", aliceConn, MID 3)+  bob <# ("", aliceConn, SENT 3)+  alice <#= \case ("", c, Msg "hello too") -> c == bobConn; _ -> False+  alice #: ("3a", bobConn, "ACK 3") #> ("3a", bobConn, OK)+  bob #: ("15", aliceConn, "SEND 9\nmessage 1") #> ("15", aliceConn, MID 4)+  bob <# ("", aliceConn, SENT 4)+  alice <#= \case ("", c, Msg "message 1") -> c == bobConn; _ -> False+  alice #: ("4a", bobConn, "ACK 4") #> ("4a", bobConn, OK)+  alice #: ("5", bobConn, "OFF") #> ("5", bobConn, OK)+  bob #: ("17", aliceConn, "SEND 9\nmessage 3") #> ("17", aliceConn, MID 5)+  bob <# ("", aliceConn, MERR 5 (SMP AUTH))+  alice #: ("6", bobConn, "DEL") #> ("6", bobConn, OK)+  alice #:# "nothing else should be delivered to alice"+ testSubscription :: Transport c => TProxy c -> c -> c -> c -> IO () testSubscription _ alice1 alice2 bob = do-  ("1", "bob", Right (INV qInfo)) <- alice1 #: ("1", "bob", "NEW")-  let qInfo' = serializeSmpQueueInfo qInfo-  bob #: ("11", "alice", "JOIN " <> qInfo') #> ("", "alice", CON)-  bob #: ("12", "alice", "SEND 5\nhello") =#> \case ("12", "alice", SENT _) -> True; _ -> False-  bob #: ("13", "alice", "SEND 11\nhello again") =#> \case ("13", "alice", SENT _) -> True; _ -> False-  alice1 <# ("", "bob", CON)+  (alice1, "alice") `connect` (bob, "bob")+  bob #: ("12", "alice", "SEND 5\nhello") #> ("12", "alice", MID 1)+  bob <# ("", "alice", SENT 1)   alice1 <#= \case ("", "bob", Msg "hello") -> True; _ -> False+  alice1 #: ("1", "bob", "ACK 1") #> ("1", "bob", OK)+  bob #: ("13", "alice", "SEND 11\nhello again") #> ("13", "alice", MID 2)+  bob <# ("", "alice", SENT 2)   alice1 <#= \case ("", "bob", Msg "hello again") -> True; _ -> False+  alice1 #: ("2", "bob", "ACK 2") #> ("2", "bob", OK)   alice2 #: ("21", "bob", "SUB") #> ("21", "bob", OK)   alice1 <# ("", "bob", END)-  bob #: ("14", "alice", "SEND 2\nhi") =#> \case ("14", "alice", SENT _) -> True; _ -> False+  bob #: ("14", "alice", "SEND 2\nhi") #> ("14", "alice", MID 3)+  bob <# ("", "alice", SENT 3)   alice2 <#= \case ("", "bob", Msg "hi") -> True; _ -> False+  alice2 #: ("22", "bob", "ACK 3") #> ("22", "bob", OK)   alice1 #:# "nothing else should be delivered to alice1"  testSubscrNotification :: Transport c => TProxy c -> (ThreadId, ThreadId) -> c -> IO ()-testSubscrNotification _ (server, _) client = do-  client #: ("1", "conn1", "NEW") =#> \case ("1", "conn1", INV _) -> True; _ -> False+testSubscrNotification t (server, _) client = do+  client #: ("1", "conn1", "NEW") =#> \case ("1", "conn1", INV {}) -> True; _ -> False   client #:# "nothing should be delivered to client before the server is killed"   killThread server-  client <# ("", "conn1", END)+  client <# ("", "conn1", DOWN)+  withSmpServer (ATransport t) $+    client <# ("", "conn1", ERR (SMP AUTH)) -- this new server does not have the queue +testMsgDeliveryServerRestart :: Transport c => TProxy c -> c -> c -> IO ()+testMsgDeliveryServerRestart t alice bob = do+  withServer $ do+    connect (alice, "alice") (bob, "bob")+    bob #: ("1", "alice", "SEND 2\nhi") #> ("1", "alice", MID 1)+    bob <# ("", "alice", SENT 1)+    alice <#= \case ("", "bob", Msg "hi") -> True; _ -> False+    alice #: ("11", "bob", "ACK 1") #> ("11", "bob", OK)+    alice #:# "nothing else delivered before the server is killed"++  alice <# ("", "bob", DOWN)+  bob #: ("2", "alice", "SEND 11\nhello again") #> ("2", "alice", MID 2)+  bob #:# "nothing else delivered before the server is restarted"+  alice #:# "nothing else delivered before the server is restarted"++  withServer $ do+    bob <# ("", "alice", SENT 2)+    alice <# ("", "bob", UP)+    alice <#= \case ("", "bob", Msg "hello again") -> True; _ -> False+    alice #: ("12", "bob", "ACK 2") #> ("12", "bob", OK)++  removeFile testStoreLogFile+  where+    withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()++testMsgDeliveryAgentRestart :: Transport c => TProxy c -> c -> IO ()+testMsgDeliveryAgentRestart t bob = do+  withAgent $ \alice -> do+    withServer $ do+      connect (bob, "bob") (alice, "alice")+      alice #: ("1", "bob", "SEND 5\nhello") #> ("1", "bob", MID 1)+      alice <# ("", "bob", SENT 1)+      bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False+      bob #: ("11", "alice", "ACK 1") #> ("11", "alice", OK)+      bob #:# "nothing else delivered before the server is down"++    bob <# ("", "alice", DOWN)+    alice #: ("2", "bob", "SEND 11\nhello again") #> ("2", "bob", MID 2)+    alice #:# "nothing else delivered before the server is restarted"+    bob #:# "nothing else delivered before the server is restarted"++  withAgent $ \alice -> do+    withServer $ do+      tPutRaw alice ("3", "bob", "SUB")+      alice <#= \case+        (corrId, "bob", cmd) ->+          (corrId == "3" && cmd == OK)+            || (corrId == "" && cmd == SENT 2)+        _ -> False+      bob <# ("", "alice", UP)+      bob <#= \case ("", "alice", Msg "hello again") -> True; _ -> False+      bob #: ("12", "alice", "ACK 2") #> ("12", "alice", OK)++  removeFile testStoreLogFile+  removeFile testDB+  where+    withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()+    withAgent = withSmpAgentThreadOn_ (ATransport t) (agentTestPort, testPort, testDB) (pure ()) . const . testSMPAgentClientOn agentTestPort++connect :: forall c. Transport c => (c, ByteString) -> (c, ByteString) -> IO ()+connect (h1, name1) (h2, name2) = do+  ("c1", _, Right (INV qInfo)) <- h1 #: ("c1", name2, "NEW")+  let qInfo' = serializeSmpQueueInfo qInfo+  h2 #: ("c2", name1, "JOIN " <> qInfo' <> " 5\ninfo2") #> ("c2", name1, OK)+  ("", _, Right (REQ connId "info2")) <- (h1 <#:)+  h1 #: ("c3", name2, "ACPT " <> connId <> " 5\ninfo1") #> ("c3", name2, OK)+  h2 <# ("", name1, INFO "info1")+  h2 <# ("", name1, CON)+  h1 <# ("", name2, CON)++-- connect' :: forall c. Transport c => c -> c -> IO (ByteString, ByteString)+-- connect' h1 h2 = do+--   ("c1", conn2, Right (INV qInfo)) <- h1 #: ("c1", "", "NEW")+--   let qInfo' = serializeSmpQueueInfo qInfo+--   ("c2", conn1, Right OK) <- h2 #: ("c2", "", "JOIN " <> qInfo' <> " 5\ninfo2")+--   ("", _, Right (REQ connId "info2")) <- (h1 <#:)+--   h1 #: ("c3", conn2, "ACPT " <> connId <> " 5\ninfo1") =#> \case ("c3", c, OK) -> c == conn2; _ -> False+--   h2 <# ("", conn1, INFO "info1")+--   h2 <# ("", conn1, CON)+--   h1 <# ("", conn2, CON)+--   pure (conn1, conn2)+ samplePublicKey :: ByteString samplePublicKey = "rsa:MIIBoDANBgkqhkiG9w0BAQEFAAOCAY0AMIIBiAKCAQEAtn1NI2tPoOGSGfad0aUg0tJ0kG2nzrIPGLiz8wb3dQSJC9xkRHyzHhEE8Kmy2cM4q7rNZIlLcm4M7oXOTe7SC4x59bLQG9bteZPKqXu9wk41hNamV25PWQ4zIcIRmZKETVGbwN7jFMpH7wxLdI1zzMArAPKXCDCJ5ctWh4OWDI6OR6AcCtEj+toCI6N6pjxxn5VigJtwiKhxYpoUJSdNM60wVEDCSUrZYBAuDH8pOxPfP+Tm4sokaFDTIG3QJFzOjC+/9nW4MUjAOFll9PCp9kaEFHJ/YmOYKMWNOCCPvLS6lxA83i0UaardkNLNoFS5paWfTlroxRwOC2T6PwO2ywKBgDjtXcSED61zK1seocQMyGRINnlWdhceD669kIHju/f6kAayvYKW3/lbJNXCmyinAccBosO08/0sUxvtuniIo18kfYJE0UmP1ReCjhMP+O+yOmwZJini/QelJk/Pez8IIDDWnY1qYQsN/q7ocjakOYrpGG7mig6JMFpDJtD6istR" @@ -126,9 +279,8 @@   it "unknown command" $ ("1", "5678", "HELLO") >#> ("1", "5678", "ERR CMD SYNTAX")   describe "NEW" do     describe "valid" do-      -- TODO: ERROR no connection alias in the response (it does not generate it yet if not provided)       -- TODO: add tests with defined connection alias-      xit "without parameters" $ ("211", "", "NEW") >#>= \case ("211", "", "INV" : _) -> True; _ -> False+      it "without parameters" $ ("211", "", "NEW") >#>= \case ("211", _, "INV" : _) -> True; _ -> False     describe "invalid" do       -- TODO: add tests with defined connection alias       it "with parameters" $ ("222", "", "NEW hi") >#> ("222", "", "ERR CMD SYNTAX")@@ -138,7 +290,7 @@       -- TODO: ERROR no connection alias in the response (it does not generate it yet if not provided)       -- TODO: add tests with defined connection alias       it "using same server as in invitation" $-        ("311", "", "JOIN smp::localhost:5000::1234::" <> samplePublicKey) >#> ("311", "", "ERR SMP AUTH")+        ("311", "a", "JOIN smp::localhost:5000::1234::" <> samplePublicKey <> " 14\nbob's connInfo") >#> ("311", "a", "ERR SMP AUTH")     describe "invalid" do       -- TODO: JOIN is not merged yet - to be added       it "no parameters" $ ("321", "", "JOIN") >#> ("321", "", "ERR CMD SYNTAX")
+ tests/AgentTests/FunctionalAPITests.hs view
@@ -0,0 +1,163 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE PatternSynonyms #-}+{-# OPTIONS_GHC -fno-warn-incomplete-uni-patterns #-}++module AgentTests.FunctionalAPITests (functionalAPITests) where++import Control.Monad.Except (ExceptT, runExceptT)+import Control.Monad.IO.Unlift+import SMPAgentClient+import SMPClient (withSmpServer)+import Simplex.Messaging.Agent+import Simplex.Messaging.Agent.Env.SQLite (dbFile)+import Simplex.Messaging.Agent.Protocol+import Simplex.Messaging.Protocol (ErrorType (..), MsgBody)+import Simplex.Messaging.Transport (ATransport (..))+import System.Timeout+import Test.Hspec+import UnliftIO.STM++(##>) :: MonadIO m => m (ATransmission 'Agent) -> ATransmission 'Agent -> m ()+a ##> t = a >>= \t' -> liftIO (t' `shouldBe` t)++(=##>) :: MonadIO m => m (ATransmission 'Agent) -> (ATransmission 'Agent -> Bool) -> m ()+a =##> p = a >>= \t -> liftIO (t `shouldSatisfy` p)++get :: MonadIO m => AgentClient -> m (ATransmission 'Agent)+get c = atomically (readTBQueue $ subQ c)++pattern Msg :: MsgBody -> ACommand 'Agent+pattern Msg msgBody <- MSG MsgMeta {integrity = MsgOk} msgBody++functionalAPITests :: ATransport -> Spec+functionalAPITests t = do+  describe "Establishing duplex connection" $+    it "should connect via one server using SMP agent clients" $+      withSmpServer t testAgentClient+  describe "Establishing connection asynchronously" $ do+    it "should connect with initiating client going offline" $+      withSmpServer t testAsyncInitiatingOffline+    it "should connect with joining client going offline before its queue activation" $+      withSmpServer t testAsyncJoiningOfflineBeforeActivation+    -- TODO a valid test case but not trivial to implement, probably requires some agent rework+    xit "should connect with joining client going offline after its queue activation" $+      withSmpServer t testAsyncJoiningOfflineAfterActivation+    it "should connect with both clients going offline" $+      withSmpServer t testAsyncBothOffline++testAgentClient :: IO ()+testAgentClient = do+  alice <- getSMPAgentClient cfg+  bob <- getSMPAgentClient cfg {dbFile = testDB2}+  Right () <- runExceptT $ do+    (bobId, qInfo) <- createConnection alice+    aliceId <- joinConnection bob qInfo "bob's connInfo"+    ("", _, REQ confId "bob's connInfo") <- get alice+    acceptConnection alice bobId confId "alice's connInfo"+    get alice ##> ("", bobId, CON)+    get bob ##> ("", aliceId, INFO "alice's connInfo")+    get bob ##> ("", aliceId, CON)+    1 <- sendMessage alice bobId "hello"+    get alice ##> ("", bobId, SENT 1)+    2 <- sendMessage alice bobId "how are you?"+    get alice ##> ("", bobId, SENT 2)+    get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False+    ackMessage bob aliceId 1+    get bob =##> \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False+    ackMessage bob aliceId 2+    3 <- sendMessage bob aliceId "hello too"+    get bob ##> ("", aliceId, SENT 3)+    4 <- sendMessage bob aliceId "message 1"+    get bob ##> ("", aliceId, SENT 4)+    get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False+    ackMessage alice bobId 3+    get alice =##> \case ("", c, Msg "message 1") -> c == bobId; _ -> False+    ackMessage alice bobId 4+    suspendConnection alice bobId+    5 <- sendMessage bob aliceId "message 2"+    get bob ##> ("", aliceId, MERR 5 (SMP AUTH))+    deleteConnection alice bobId+    liftIO $ noMessages alice "nothing else should be delivered to alice"+  pure ()+  where+    noMessages :: AgentClient -> String -> Expectation+    noMessages c err = tryGet `shouldReturn` ()+      where+        tryGet =+          10000 `timeout` get c >>= \case+            Just _ -> error err+            _ -> return ()++testAsyncInitiatingOffline :: IO ()+testAsyncInitiatingOffline = do+  alice <- getSMPAgentClient cfg+  bob <- getSMPAgentClient cfg {dbFile = testDB2}+  Right () <- runExceptT $ do+    (bobId, qInfo) <- createConnection alice+    disconnectAgentClient alice+    aliceId <- joinConnection bob qInfo "bob's connInfo"+    alice' <- liftIO $ getSMPAgentClient cfg+    subscribeConnection alice' bobId+    ("", _, REQ confId "bob's connInfo") <- get alice'+    acceptConnection alice' bobId confId "alice's connInfo"+    get alice' ##> ("", bobId, CON)+    get bob ##> ("", aliceId, INFO "alice's connInfo")+    get bob ##> ("", aliceId, CON)+    exchangeGreetings alice' bobId bob aliceId+  pure ()++testAsyncJoiningOfflineBeforeActivation :: IO ()+testAsyncJoiningOfflineBeforeActivation = do+  alice <- getSMPAgentClient cfg+  bob <- getSMPAgentClient cfg {dbFile = testDB2}+  Right () <- runExceptT $ do+    (bobId, qInfo) <- createConnection alice+    aliceId <- joinConnection bob qInfo "bob's connInfo"+    disconnectAgentClient bob+    ("", _, REQ confId "bob's connInfo") <- get alice+    acceptConnection alice bobId confId "alice's connInfo"+    bob' <- liftIO $ getSMPAgentClient cfg {dbFile = testDB2}+    subscribeConnection bob' aliceId+    get alice ##> ("", bobId, CON)+    get bob' ##> ("", aliceId, INFO "alice's connInfo")+    get bob' ##> ("", aliceId, CON)+    exchangeGreetings alice bobId bob' aliceId+  pure ()++testAsyncJoiningOfflineAfterActivation :: IO ()+testAsyncJoiningOfflineAfterActivation = error "not implemented"++testAsyncBothOffline :: IO ()+testAsyncBothOffline = do+  alice <- getSMPAgentClient cfg+  bob <- getSMPAgentClient cfg {dbFile = testDB2}+  Right () <- runExceptT $ do+    (bobId, qInfo) <- createConnection alice+    disconnectAgentClient alice+    aliceId <- joinConnection bob qInfo "bob's connInfo"+    disconnectAgentClient bob+    alice' <- liftIO $ getSMPAgentClient cfg+    subscribeConnection alice' bobId+    ("", _, REQ confId "bob's connInfo") <- get alice'+    acceptConnection alice' bobId confId "alice's connInfo"+    bob' <- liftIO $ getSMPAgentClient cfg {dbFile = testDB2}+    subscribeConnection bob' aliceId+    get alice' ##> ("", bobId, CON)+    get bob' ##> ("", aliceId, INFO "alice's connInfo")+    get bob' ##> ("", aliceId, CON)+    exchangeGreetings alice' bobId bob' aliceId+  pure ()++exchangeGreetings :: AgentClient -> ConnId -> AgentClient -> ConnId -> ExceptT AgentErrorType IO ()+exchangeGreetings alice bobId bob aliceId = do+  1 <- sendMessage alice bobId "hello"+  get alice ##> ("", bobId, SENT 1)+  get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False+  ackMessage bob aliceId 1+  2 <- sendMessage bob aliceId "hello too"+  get bob ##> ("", aliceId, SENT 2)+  get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False+  ackMessage alice bobId 2
tests/AgentTests/SQLiteTests.hs view
@@ -1,5 +1,6 @@ {-# LANGUAGE BlockArguments #-} {-# LANGUAGE DuplicateRecordFields #-}+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE QuasiQuotes #-}@@ -8,9 +9,11 @@ module AgentTests.SQLiteTests (storeTests) where  import Control.Concurrent.Async (concurrently_)+import Control.Concurrent.STM import Control.Monad (replicateM_) import Control.Monad.Except (ExceptT, runExceptT) import qualified Crypto.PubKey.RSA as R+import Crypto.Random (drgNew) import Data.ByteString.Char8 (ByteString) import qualified Data.Text as T import Data.Text.Encoding (encodeUtf8)@@ -22,8 +25,9 @@ import Simplex.Messaging.Agent.Protocol import Simplex.Messaging.Agent.Store import Simplex.Messaging.Agent.Store.SQLite+import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations import qualified Simplex.Messaging.Crypto as C-import System.Random (Random (randomIO))+import System.Random import Test.Hspec import UnliftIO.Directory (removeFile) @@ -39,7 +43,7 @@     connect2 :: IO (SQLiteStore, SQLiteStore)     connect2 = do       s1 <- createStore-      s2 <- connectSQLiteStore $ dbFilePath s1+      s2 <- connectSQLiteStore (dbFilePath s1) 4       pure (s1, s2)  createStore :: IO SQLiteStore@@ -47,12 +51,15 @@   -- Randomize DB file name to avoid SQLite IO errors supposedly caused by asynchronous   -- IO operations on multiple similarly named files; error seems to be environment specific   r <- randomIO :: IO Word32-  createSQLiteStore $ testDB <> show r+  createSQLiteStore (testDB <> show r) 4 Migrations.app  removeStore :: SQLiteStore -> IO () removeStore store = do-  DB.close $ dbConn store+  close store   removeFile $ dbFilePath store+  where+    close :: SQLiteStore -> IO ()+    close st = mapM_ DB.close =<< atomically (flushTBQueue $ dbConnPool st)  returnsResult :: (Eq a, Eq e, Show a, Show e) => ExceptT e IO a -> a -> Expectation action `returnsResult` r = runExceptT action `shouldReturn` Right r@@ -73,11 +80,13 @@       describe "Queue and Connection management" do         describe "createRcvConn" do           testCreateRcvConn+          testCreateRcvConnRandomId           testCreateRcvConnDuplicate         describe "createSndConn" do           testCreateSndConn+          testCreateSndConnRandomID           testCreateSndConnDuplicate-        describe "getAllConnAliases" testGetAllConnAliases+        describe "getAllConnIds" testGetAllConnIds         describe "getRcvConn" testGetRcvConn         describe "deleteConn" do           testDeleteRcvConn@@ -104,24 +113,29 @@ testConcurrentWrites :: SpecWith (SQLiteStore, SQLiteStore) testConcurrentWrites =   it "should complete multiple concurrent write transactions w/t sqlite busy errors" $ \(s1, s2) -> do-    _ <- runExceptT $ createRcvConn s1 rcvQueue1-    concurrently_ (runTest s1) (runTest s2)+    g <- newTVarIO =<< drgNew+    _ <- runExceptT $ createRcvConn s1 g cData1 rcvQueue1+    let ConnData {connId} = cData1+    concurrently_ (runTest s1 connId) (runTest s2 connId)   where-    runTest :: SQLiteStore -> IO (Either StoreError ())-    runTest store = runExceptT . replicateM_ 100 $ do-      (internalId, internalRcvId, _, _) <- updateRcvIds store rcvQueue1+    runTest :: SQLiteStore -> ConnId -> IO (Either StoreError ())+    runTest store connId = runExceptT . replicateM_ 100 $ do+      (internalId, internalRcvId, _, _) <- updateRcvIds store connId       let rcvMsgData = mkRcvMsgData internalId internalRcvId 0 "0" "hash_dummy"-      createRcvMsg store rcvQueue1 rcvMsgData+      createRcvMsg store connId rcvMsgData  testCompiledThreadsafe :: SpecWith SQLiteStore testCompiledThreadsafe =-  it "compiled sqlite library should be threadsafe" $ \store -> do-    compileOptions <- DB.query_ (dbConn store) "pragma COMPILE_OPTIONS;" :: IO [[T.Text]]+  it "compiled sqlite library should be threadsafe" . withStoreConnection $ \db -> do+    compileOptions <- DB.query_ db "pragma COMPILE_OPTIONS;" :: IO [[T.Text]]     compileOptions `shouldNotContain` [["THREADSAFE=0"]] +withStoreConnection :: (DB.Connection -> IO a) -> SQLiteStore -> IO a+withStoreConnection = flip withConnection+ testForeignKeysEnabled :: SpecWith SQLiteStore testForeignKeysEnabled =-  it "foreign keys should be enabled" $ \store -> do+  it "foreign keys should be enabled" . withStoreConnection $ \db -> do     let inconsistentQuery =           [sql|             INSERT INTO connections@@ -129,18 +143,19 @@             VALUES               ("conn1", "smp.simplex.im", "5223", "1234", "smp.simplex.im", "5223", "2345");           |]-    DB.execute_ (dbConn store) inconsistentQuery+    DB.execute_ db inconsistentQuery       `shouldThrow` (\e -> DB.sqlError e == DB.ErrorConstraint) +cData1 :: ConnData+cData1 = ConnData {connId = "conn1"}+ rcvQueue1 :: RcvQueue rcvQueue1 =   RcvQueue     { server = SMPServer "smp.simplex.im" (Just "5223") testKeyHash,       rcvId = "1234",-      connAlias = "conn1",       rcvPrivateKey = C.safePrivateKey (1, 2, 3),       sndId = Just "2345",-      sndKey = Nothing,       decryptKey = C.safePrivateKey (1, 2, 3),       verifyKey = Nothing,       status = New@@ -151,74 +166,104 @@   SndQueue     { server = SMPServer "smp.simplex.im" (Just "5223") testKeyHash,       sndId = "3456",-      connAlias = "conn1",       sndPrivateKey = C.safePrivateKey (1, 2, 3),       encryptKey = C.PublicKey $ R.PublicKey 1 2 3,-      signKey = C.safePrivateKey (1, 2, 3),+      signKey = C.APrivateKey $ C.unPrivateKey (C.safePrivateKey (1, 2, 3) :: C.SafePrivateKey),       status = New     }  testCreateRcvConn :: SpecWith SQLiteStore testCreateRcvConn =   it "should create RcvConnection and add SndQueue" $ \store -> do-    createRcvConn store rcvQueue1-      `returnsResult` ()+    g <- newTVarIO =<< drgNew+    createRcvConn store g cData1 rcvQueue1+      `returnsResult` "conn1"     getConn store "conn1"-      `returnsResult` SomeConn SCRcv (RcvConnection "conn1" rcvQueue1)+      `returnsResult` SomeConn SCRcv (RcvConnection cData1 rcvQueue1)     upgradeRcvConnToDuplex store "conn1" sndQueue1       `returnsResult` ()     getConn store "conn1"-      `returnsResult` SomeConn SCDuplex (DuplexConnection "conn1" rcvQueue1 sndQueue1)+      `returnsResult` SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1) +testCreateRcvConnRandomId :: SpecWith SQLiteStore+testCreateRcvConnRandomId =+  it "should create RcvConnection and add SndQueue with random ID" $ \store -> do+    g <- newTVarIO =<< drgNew+    Right connId <- runExceptT $ createRcvConn store g cData1 {connId = ""} rcvQueue1+    getConn store connId+      `returnsResult` SomeConn SCRcv (RcvConnection cData1 {connId} rcvQueue1)+    upgradeRcvConnToDuplex store connId sndQueue1+      `returnsResult` ()+    getConn store connId+      `returnsResult` SomeConn SCDuplex (DuplexConnection cData1 {connId} rcvQueue1 sndQueue1)+ testCreateRcvConnDuplicate :: SpecWith SQLiteStore testCreateRcvConnDuplicate =   it "should throw error on attempt to create duplicate RcvConnection" $ \store -> do-    _ <- runExceptT $ createRcvConn store rcvQueue1-    createRcvConn store rcvQueue1+    g <- newTVarIO =<< drgNew+    _ <- runExceptT $ createRcvConn store g cData1 rcvQueue1+    createRcvConn store g cData1 rcvQueue1       `throwsError` SEConnDuplicate  testCreateSndConn :: SpecWith SQLiteStore testCreateSndConn =   it "should create SndConnection and add RcvQueue" $ \store -> do-    createSndConn store sndQueue1-      `returnsResult` ()+    g <- newTVarIO =<< drgNew+    createSndConn store g cData1 sndQueue1+      `returnsResult` "conn1"     getConn store "conn1"-      `returnsResult` SomeConn SCSnd (SndConnection "conn1" sndQueue1)+      `returnsResult` SomeConn SCSnd (SndConnection cData1 sndQueue1)     upgradeSndConnToDuplex store "conn1" rcvQueue1       `returnsResult` ()     getConn store "conn1"-      `returnsResult` SomeConn SCDuplex (DuplexConnection "conn1" rcvQueue1 sndQueue1)+      `returnsResult` SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1) +testCreateSndConnRandomID :: SpecWith SQLiteStore+testCreateSndConnRandomID =+  it "should create SndConnection and add RcvQueue with random ID" $ \store -> do+    g <- newTVarIO =<< drgNew+    Right connId <- runExceptT $ createSndConn store g cData1 {connId = ""} sndQueue1+    getConn store connId+      `returnsResult` SomeConn SCSnd (SndConnection cData1 {connId} sndQueue1)+    upgradeSndConnToDuplex store connId rcvQueue1+      `returnsResult` ()+    getConn store connId+      `returnsResult` SomeConn SCDuplex (DuplexConnection cData1 {connId} rcvQueue1 sndQueue1)+ testCreateSndConnDuplicate :: SpecWith SQLiteStore testCreateSndConnDuplicate =   it "should throw error on attempt to create duplicate SndConnection" $ \store -> do-    _ <- runExceptT $ createSndConn store sndQueue1-    createSndConn store sndQueue1+    g <- newTVarIO =<< drgNew+    _ <- runExceptT $ createSndConn store g cData1 sndQueue1+    createSndConn store g cData1 sndQueue1       `throwsError` SEConnDuplicate -testGetAllConnAliases :: SpecWith SQLiteStore-testGetAllConnAliases =+testGetAllConnIds :: SpecWith SQLiteStore+testGetAllConnIds =   it "should get all conn aliases" $ \store -> do-    _ <- runExceptT $ createRcvConn store rcvQueue1-    _ <- runExceptT $ createSndConn store sndQueue1 {connAlias = "conn2"}-    getAllConnAliases store-      `returnsResult` ["conn1" :: ConnAlias, "conn2" :: ConnAlias]+    g <- newTVarIO =<< drgNew+    _ <- runExceptT $ createRcvConn store g cData1 rcvQueue1+    _ <- runExceptT $ createSndConn store g cData1 {connId = "conn2"} sndQueue1+    getAllConnIds store+      `returnsResult` ["conn1" :: ConnId, "conn2" :: ConnId]  testGetRcvConn :: SpecWith SQLiteStore testGetRcvConn =   it "should get connection using rcv queue id and server" $ \store -> do     let smpServer = SMPServer "smp.simplex.im" (Just "5223") testKeyHash     let recipientId = "1234"-    _ <- runExceptT $ createRcvConn store rcvQueue1+    g <- newTVarIO =<< drgNew+    _ <- runExceptT $ createRcvConn store g cData1 rcvQueue1     getRcvConn store smpServer recipientId-      `returnsResult` SomeConn SCRcv (RcvConnection (connAlias (rcvQueue1 :: RcvQueue)) rcvQueue1)+      `returnsResult` SomeConn SCRcv (RcvConnection cData1 rcvQueue1)  testDeleteRcvConn :: SpecWith SQLiteStore testDeleteRcvConn =   it "should create RcvConnection and delete it" $ \store -> do-    _ <- runExceptT $ createRcvConn store rcvQueue1+    g <- newTVarIO =<< drgNew+    _ <- runExceptT $ createRcvConn store g cData1 rcvQueue1     getConn store "conn1"-      `returnsResult` SomeConn SCRcv (RcvConnection "conn1" rcvQueue1)+      `returnsResult` SomeConn SCRcv (RcvConnection cData1 rcvQueue1)     deleteConn store "conn1"       `returnsResult` ()     -- TODO check queues are deleted as well@@ -228,9 +273,10 @@ testDeleteSndConn :: SpecWith SQLiteStore testDeleteSndConn =   it "should create SndConnection and delete it" $ \store -> do-    _ <- runExceptT $ createSndConn store sndQueue1+    g <- newTVarIO =<< drgNew+    _ <- runExceptT $ createSndConn store g cData1 sndQueue1     getConn store "conn1"-      `returnsResult` SomeConn SCSnd (SndConnection "conn1" sndQueue1)+      `returnsResult` SomeConn SCSnd (SndConnection cData1 sndQueue1)     deleteConn store "conn1"       `returnsResult` ()     -- TODO check queues are deleted as well@@ -240,10 +286,11 @@ testDeleteDuplexConn :: SpecWith SQLiteStore testDeleteDuplexConn =   it "should create DuplexConnection and delete it" $ \store -> do-    _ <- runExceptT $ createRcvConn store rcvQueue1+    g <- newTVarIO =<< drgNew+    _ <- runExceptT $ createRcvConn store g cData1 rcvQueue1     _ <- runExceptT $ upgradeRcvConnToDuplex store "conn1" sndQueue1     getConn store "conn1"-      `returnsResult` SomeConn SCDuplex (DuplexConnection "conn1" rcvQueue1 sndQueue1)+      `returnsResult` SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1)     deleteConn store "conn1"       `returnsResult` ()     -- TODO check queues are deleted as well@@ -253,15 +300,15 @@ testUpgradeRcvConnToDuplex :: SpecWith SQLiteStore testUpgradeRcvConnToDuplex =   it "should throw error on attempt to add SndQueue to SndConnection or DuplexConnection" $ \store -> do-    _ <- runExceptT $ createSndConn store sndQueue1+    g <- newTVarIO =<< drgNew+    _ <- runExceptT $ createSndConn store g cData1 sndQueue1     let anotherSndQueue =           SndQueue             { server = SMPServer "smp.simplex.im" (Just "5223") testKeyHash,               sndId = "2345",-              connAlias = "conn1",               sndPrivateKey = C.safePrivateKey (1, 2, 3),               encryptKey = C.PublicKey $ R.PublicKey 1 2 3,-              signKey = C.safePrivateKey (1, 2, 3),+              signKey = C.APrivateKey $ C.unPrivateKey (C.safePrivateKey (1, 2, 3) :: C.SafePrivateKey),               status = New             }     upgradeRcvConnToDuplex store "conn1" anotherSndQueue@@ -273,15 +320,14 @@ testUpgradeSndConnToDuplex :: SpecWith SQLiteStore testUpgradeSndConnToDuplex =   it "should throw error on attempt to add RcvQueue to RcvConnection or DuplexConnection" $ \store -> do-    _ <- runExceptT $ createRcvConn store rcvQueue1+    g <- newTVarIO =<< drgNew+    _ <- runExceptT $ createRcvConn store g cData1 rcvQueue1     let anotherRcvQueue =           RcvQueue             { server = SMPServer "smp.simplex.im" (Just "5223") testKeyHash,               rcvId = "3456",-              connAlias = "conn1",               rcvPrivateKey = C.safePrivateKey (1, 2, 3),               sndId = Just "4567",-              sndKey = Nothing,               decryptKey = C.safePrivateKey (1, 2, 3),               verifyKey = Nothing,               status = New@@ -295,40 +341,43 @@ testSetRcvQueueStatus :: SpecWith SQLiteStore testSetRcvQueueStatus =   it "should update status of RcvQueue" $ \store -> do-    _ <- runExceptT $ createRcvConn store rcvQueue1+    g <- newTVarIO =<< drgNew+    _ <- runExceptT $ createRcvConn store g cData1 rcvQueue1     getConn store "conn1"-      `returnsResult` SomeConn SCRcv (RcvConnection "conn1" rcvQueue1)+      `returnsResult` SomeConn SCRcv (RcvConnection cData1 rcvQueue1)     setRcvQueueStatus store rcvQueue1 Confirmed       `returnsResult` ()     getConn store "conn1"-      `returnsResult` SomeConn SCRcv (RcvConnection "conn1" rcvQueue1 {status = Confirmed})+      `returnsResult` SomeConn SCRcv (RcvConnection cData1 rcvQueue1 {status = Confirmed})  testSetSndQueueStatus :: SpecWith SQLiteStore testSetSndQueueStatus =   it "should update status of SndQueue" $ \store -> do-    _ <- runExceptT $ createSndConn store sndQueue1+    g <- newTVarIO =<< drgNew+    _ <- runExceptT $ createSndConn store g cData1 sndQueue1     getConn store "conn1"-      `returnsResult` SomeConn SCSnd (SndConnection "conn1" sndQueue1)+      `returnsResult` SomeConn SCSnd (SndConnection cData1 sndQueue1)     setSndQueueStatus store sndQueue1 Confirmed       `returnsResult` ()     getConn store "conn1"-      `returnsResult` SomeConn SCSnd (SndConnection "conn1" sndQueue1 {status = Confirmed})+      `returnsResult` SomeConn SCSnd (SndConnection cData1 sndQueue1 {status = Confirmed})  testSetQueueStatusDuplex :: SpecWith SQLiteStore testSetQueueStatusDuplex =   it "should update statuses of RcvQueue and SndQueue in DuplexConnection" $ \store -> do-    _ <- runExceptT $ createRcvConn store rcvQueue1+    g <- newTVarIO =<< drgNew+    _ <- runExceptT $ createRcvConn store g cData1 rcvQueue1     _ <- runExceptT $ upgradeRcvConnToDuplex store "conn1" sndQueue1     getConn store "conn1"-      `returnsResult` SomeConn SCDuplex (DuplexConnection "conn1" rcvQueue1 sndQueue1)+      `returnsResult` SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1)     setRcvQueueStatus store rcvQueue1 Secured       `returnsResult` ()     getConn store "conn1"-      `returnsResult` SomeConn SCDuplex (DuplexConnection "conn1" rcvQueue1 {status = Secured} sndQueue1)+      `returnsResult` SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 {status = Secured} sndQueue1)     setSndQueueStatus store sndQueue1 Confirmed       `returnsResult` ()     getConn store "conn1"-      `returnsResult` SomeConn SCDuplex (DuplexConnection "conn1" rcvQueue1 {status = Secured} sndQueue1 {status = Confirmed})+      `returnsResult` SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 {status = Secured} sndQueue1 {status = Confirmed})  testSetRcvQueueStatusNoQueue :: SpecWith SQLiteStore testSetRcvQueueStatusNoQueue =@@ -351,31 +400,36 @@ mkRcvMsgData :: InternalId -> InternalRcvId -> ExternalSndId -> BrokerId -> MsgHash -> RcvMsgData mkRcvMsgData internalId internalRcvId externalSndId brokerId internalHash =   RcvMsgData-    { internalId,-      internalRcvId,-      internalTs = ts,-      senderMeta = (externalSndId, ts),-      brokerMeta = (brokerId, ts),+    { internalRcvId,+      msgMeta =+        MsgMeta+          { integrity = MsgOk,+            recipient = (unId internalId, ts),+            sender = (externalSndId, ts),+            broker = (brokerId, ts)+          },       msgBody = hw,       internalHash,-      externalPrevSndHash = "hash_from_sender",-      msgIntegrity = MsgOk+      externalPrevSndHash = "hash_from_sender"     } -testCreateRcvMsg' :: SQLiteStore -> PrevExternalSndId -> PrevRcvMsgHash -> RcvQueue -> RcvMsgData -> Expectation-testCreateRcvMsg' store expectedPrevSndId expectedPrevHash rcvQueue rcvMsgData@RcvMsgData {..} = do-  updateRcvIds store rcvQueue-    `returnsResult` (internalId, internalRcvId, expectedPrevSndId, expectedPrevHash)-  createRcvMsg store rcvQueue rcvMsgData+testCreateRcvMsg' :: SQLiteStore -> PrevExternalSndId -> PrevRcvMsgHash -> ConnId -> RcvMsgData -> Expectation+testCreateRcvMsg' st expectedPrevSndId expectedPrevHash connId rcvMsgData@RcvMsgData {..} = do+  let MsgMeta {recipient = (internalId, _)} = msgMeta+  updateRcvIds st connId+    `returnsResult` (InternalId internalId, internalRcvId, expectedPrevSndId, expectedPrevHash)+  createRcvMsg st connId rcvMsgData     `returnsResult` ()  testCreateRcvMsg :: SpecWith SQLiteStore testCreateRcvMsg =-  it "should reserve internal ids and create a RcvMsg" $ \store -> do-    _ <- runExceptT $ createRcvConn store rcvQueue1+  it "should reserve internal ids and create a RcvMsg" $ \st -> do+    g <- newTVarIO =<< drgNew+    let ConnData {connId} = cData1+    _ <- runExceptT $ createRcvConn st g cData1 rcvQueue1     -- TODO getMsg to check message-    testCreateRcvMsg' store 0 "" rcvQueue1 $ mkRcvMsgData (InternalId 1) (InternalRcvId 1) 1 "1" "hash_dummy"-    testCreateRcvMsg' store 1 "hash_dummy" rcvQueue1 $ mkRcvMsgData (InternalId 2) (InternalRcvId 2) 2 "2" "new_hash_dummy"+    testCreateRcvMsg' st 0 "" connId $ mkRcvMsgData (InternalId 1) (InternalRcvId 1) 1 "1" "hash_dummy"+    testCreateRcvMsg' st 1 "hash_dummy" connId $ mkRcvMsgData (InternalId 2) (InternalRcvId 2) 2 "2" "new_hash_dummy"  mkSndMsgData :: InternalId -> InternalSndId -> MsgHash -> SndMsgData mkSndMsgData internalId internalSndId internalHash =@@ -384,32 +438,37 @@       internalSndId,       internalTs = ts,       msgBody = hw,-      internalHash+      internalHash,+      previousMsgHash = internalHash     } -testCreateSndMsg' :: SQLiteStore -> PrevSndMsgHash -> SndQueue -> SndMsgData -> Expectation-testCreateSndMsg' store expectedPrevHash sndQueue sndMsgData@SndMsgData {..} = do-  updateSndIds store sndQueue+testCreateSndMsg' :: SQLiteStore -> PrevSndMsgHash -> ConnId -> SndMsgData -> Expectation+testCreateSndMsg' store expectedPrevHash connId sndMsgData@SndMsgData {..} = do+  updateSndIds store connId     `returnsResult` (internalId, internalSndId, expectedPrevHash)-  createSndMsg store sndQueue sndMsgData+  createSndMsg store connId sndMsgData     `returnsResult` ()  testCreateSndMsg :: SpecWith SQLiteStore testCreateSndMsg =   it "should create a SndMsg and return InternalId and PrevSndMsgHash" $ \store -> do-    _ <- runExceptT $ createSndConn store sndQueue1+    g <- newTVarIO =<< drgNew+    let ConnData {connId} = cData1+    _ <- runExceptT $ createSndConn store g cData1 sndQueue1     -- TODO getMsg to check message-    testCreateSndMsg' store "" sndQueue1 $ mkSndMsgData (InternalId 1) (InternalSndId 1) "hash_dummy"-    testCreateSndMsg' store "hash_dummy" sndQueue1 $ mkSndMsgData (InternalId 2) (InternalSndId 2) "new_hash_dummy"+    testCreateSndMsg' store "" connId $ mkSndMsgData (InternalId 1) (InternalSndId 1) "hash_dummy"+    testCreateSndMsg' store "hash_dummy" connId $ mkSndMsgData (InternalId 2) (InternalSndId 2) "new_hash_dummy"  testCreateRcvAndSndMsgs :: SpecWith SQLiteStore testCreateRcvAndSndMsgs =   it "should create multiple RcvMsg and SndMsg, correctly ordering internal Ids and returning previous state" $ \store -> do-    _ <- runExceptT $ createRcvConn store rcvQueue1+    g <- newTVarIO =<< drgNew+    let ConnData {connId} = cData1+    _ <- runExceptT $ createRcvConn store g cData1 rcvQueue1     _ <- runExceptT $ upgradeRcvConnToDuplex store "conn1" sndQueue1-    testCreateRcvMsg' store 0 "" rcvQueue1 $ mkRcvMsgData (InternalId 1) (InternalRcvId 1) 1 "1" "rcv_hash_1"-    testCreateRcvMsg' store 1 "rcv_hash_1" rcvQueue1 $ mkRcvMsgData (InternalId 2) (InternalRcvId 2) 2 "2" "rcv_hash_2"-    testCreateSndMsg' store "" sndQueue1 $ mkSndMsgData (InternalId 3) (InternalSndId 1) "snd_hash_1"-    testCreateRcvMsg' store 2 "rcv_hash_2" rcvQueue1 $ mkRcvMsgData (InternalId 4) (InternalRcvId 3) 3 "3" "rcv_hash_3"-    testCreateSndMsg' store "snd_hash_1" sndQueue1 $ mkSndMsgData (InternalId 5) (InternalSndId 2) "snd_hash_2"-    testCreateSndMsg' store "snd_hash_2" sndQueue1 $ mkSndMsgData (InternalId 6) (InternalSndId 3) "snd_hash_3"+    testCreateRcvMsg' store 0 "" connId $ mkRcvMsgData (InternalId 1) (InternalRcvId 1) 1 "1" "rcv_hash_1"+    testCreateRcvMsg' store 1 "rcv_hash_1" connId $ mkRcvMsgData (InternalId 2) (InternalRcvId 2) 2 "2" "rcv_hash_2"+    testCreateSndMsg' store "" connId $ mkSndMsgData (InternalId 3) (InternalSndId 1) "snd_hash_1"+    testCreateRcvMsg' store 2 "rcv_hash_2" connId $ mkRcvMsgData (InternalId 4) (InternalRcvId 3) 3 "3" "rcv_hash_3"+    testCreateSndMsg' store "snd_hash_1" connId $ mkSndMsgData (InternalId 5) (InternalSndId 2) "snd_hash_2"+    testCreateSndMsg' store "snd_hash_2" connId $ mkSndMsgData (InternalId 6) (InternalSndId 3) "snd_hash_3"
tests/SMPAgentClient.hs view
@@ -24,6 +24,7 @@ import Simplex.Messaging.Agent (runSMPAgentBlocking) import Simplex.Messaging.Agent.Env.SQLite import Simplex.Messaging.Agent.Protocol+import Simplex.Messaging.Agent.RetryInterval import Simplex.Messaging.Client (SMPClientConfig (..), smpDefaultConfig) import Simplex.Messaging.Transport import Test.Hspec@@ -95,11 +96,15 @@ smpAgentTest2_2_2 :: forall c. Transport c => (c -> c -> IO ()) -> Expectation smpAgentTest2_2_2 test' =   withSmpServerOn (transport @c) testPort2 $-    smpAgentTestN-      [ (agentTestPort, testPort, testDB),-        (agentTestPort2, testPort2, testDB2)-      ]-      _test+    smpAgentTest2_2_2_needs_server test'++smpAgentTest2_2_2_needs_server :: forall c. Transport c => (c -> c -> IO ()) -> Expectation+smpAgentTest2_2_2_needs_server test' =+  smpAgentTestN+    [ (agentTestPort, testPort, testDB),+      (agentTestPort2, testPort2, testDB2)+    ]+    _test   where     _test [h1, h2] = test' h1 h2     _test _ = error "expected 2 handles"@@ -139,13 +144,20 @@     _test [h1, h2, h3] = test' h1 h2 h3     _test _ = error "expected 3 handles" +smpAgentTest1_1_1 :: forall c. Transport c => (c -> IO ()) -> Expectation+smpAgentTest1_1_1 test' =+  smpAgentTestN+    [(agentTestPort2, testPort2, testDB2)]+    _test+  where+    _test [h] = test' h+    _test _ = error "expected 1 handle"+ cfg :: AgentConfig cfg =-  AgentConfig+  defaultAgentConfig     { tcpPort = agentTestPort,       smpServers = L.fromList ["localhost:5000#KXNE1m2E1m0lm92WGKet9CL6+lO742Vy5G6nsrkvgs8="],-      rsaKeySize = 2048 `div` 8,-      connIdBytes = 12,       tbqSize = 1,       dbFile = testDB,       smpCfg =@@ -153,16 +165,20 @@           { qSize = 1,             defaultTransport = (testPort, transport @TCP),             tcpTimeout = 500_000-          }+          },+      retryInterval = (retryInterval defaultAgentConfig) {initialInterval = 50_000}     } -withSmpAgentThreadOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> (ThreadId -> m a) -> m a-withSmpAgentThreadOn t (port', smpPort', db') =+withSmpAgentThreadOn_ :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> m () -> (ThreadId -> m a) -> m a+withSmpAgentThreadOn_ t (port', smpPort', db') afterProcess =   let cfg' = cfg {tcpPort = port', dbFile = db', smpServers = L.fromList [SMPServer "localhost" (Just smpPort') testKeyHash]}    in serverBracket         (\started -> runSMPAgentBlocking t started cfg')-        (removeFile db')+        afterProcess +withSmpAgentThreadOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> (ThreadId -> m a) -> m a+withSmpAgentThreadOn t a@(_, _, db') = withSmpAgentThreadOn_ t a $ removeFile db'+ withSmpAgentOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> m a -> m a withSmpAgentOn t (port', smpPort', db') = withSmpAgentThreadOn t (port', smpPort', db') . const @@ -173,7 +189,7 @@ testSMPAgentClientOn port' client = do   runTransportClient agentTestHost port' $ \h -> do     line <- liftIO $ getLn h-    if line == "Welcome to SMP v0.3.2 agent"+    if line == "Welcome to SMP v0.4.0 agent"       then client h       else error $ "wrong welcome message: " <> B.unpack line
tests/SMPClient.hs view
@@ -39,6 +39,9 @@ testKeyHashStr :: B.ByteString testKeyHashStr = "KXNE1m2E1m0lm92WGKet9CL6+lO742Vy5G6nsrkvgs8=" +testBlockSize :: Maybe Int+testBlockSize = Just 8192+ testKeyHash :: Maybe C.KeyHash testKeyHash = Just "KXNE1m2E1m0lm92WGKet9CL6+lO742Vy5G6nsrkvgs8=" @@ -48,7 +51,7 @@ testSMPClient :: (Transport c, MonadUnliftIO m) => (THandle c -> m a) -> m a testSMPClient client =   runTransportClient testHost testPort $ \h ->-    liftIO (runExceptT $ clientHandshake h testKeyHash) >>= \case+    liftIO (runExceptT $ clientHandshake h testBlockSize testKeyHash) >>= \case       Right th -> client th       Left e -> error $ show e @@ -57,9 +60,11 @@   ServerConfig     { transports = undefined,       tbqSize = 1,+      msgQueueQuota = 4,       queueIdBytes = 12,       msgIdBytes = 6,       storeLog = Nothing,+      blockSize = 8192,       serverPrivateKey =         -- full RSA private key (only for tests)         "MIIFIwIBAAKCAQEArZyrri/NAwt5buvYjwu+B/MQeJUszDBpRgVqNddlI9kNwDXu\
tests/ServerTests.hs view
@@ -337,12 +337,19 @@         (testSameTiming rh sh)         [ (128, 128, 100),           (128, 256, 25),+          (128, 384, 15),           -- (128, 512, 15),           (256, 128, 100),-          (256, 256, 25)+          (256, 256, 25),+          (256, 384, 15),           -- (256, 512, 15),+          (384, 128, 100),+          (384, 256, 25),+          (384, 384, 15)+          -- (384, 512, 15),           -- (512, 128, 100),           -- (512, 256, 25),+          -- (512, 384, 15),           -- (512, 512, 15)         ]   where
tests/Test.hs view
@@ -1,6 +1,6 @@ {-# LANGUAGE TypeApplications #-} -import AgentTests+import AgentTests (agentTests) import ProtocolErrorTests import ServerTests import Simplex.Messaging.Transport (TCP, Transport (..))